Sanitize codebase, reorganize docs, and add missing deploy files

Remove dead code identified in docs/SANITIZATION-REVIEW.md:
- app/blueprints/content_old.py and app/blueprints/playlist.py
- app/models/group.py, app/utils/nginx_config_reader.py
- orphaned templates (content_list, edit_content, upload_content,
  player_page) and the related group/Template references

Result: 6 blueprints, 82 routes, no dead modules or orphan templates.

Add files that deploy.sh and docker-entrypoint.sh already require but
which were never tracked:
- https_manager.py       (referenced by deploy.sh, migrate_network.sh,
                          docker-entrypoint.sh)
- Caddyfile.example      (seeded by deploy.sh; its absence aborts deploy)

Relocate generated Graphify artifacts from graphify-out/ to
docs/graphify-out/ (110 files, no content change) and archive the
superseded docs under docs/.

Ignore hygiene:
- ignore ad-hoc .env backups (.env.bak*) — they contain live secrets
- keep the pre-sanitization snapshots (docs/legacy code/,
  docs/old_code_documentation/) on disk but out of the repo

Fix .env.example: drop a duplicated config block, genericize the
hardcoded host IP, and document HOSTNAME_INTERNAL.
This commit is contained in:
2026-09-11 12:18:34 +03:00
parent 1c5186463a
commit 46602f1933
226 changed files with 3999 additions and 15737 deletions
+314
View File
@@ -0,0 +1,314 @@
"""Static analyser for the digiserver-v2 sanitization pass.
Inventories every Python function/method/class in the *active* code
(app/ and migrations/), plus routes, and builds a call graph so dead and
broken code can be identified.
Outputs JSON to stdout when run with --json, otherwise a readable report.
Usage:
python tools/sanitize_audit.py
python tools/sanitize_audit.py --json > audit.json
"""
from __future__ import annotations
import argparse
import ast
import json
import os
import re
import sys
from collections import defaultdict
# Walk up until we find the project root (robust to the tool living in a
# nested folder such as docs/tools/).
_here = os.path.dirname(os.path.abspath(__file__))
while _here != os.path.dirname(_here) and not os.path.isdir(os.path.join(_here, 'app')):
_here = os.path.dirname(_here)
REPO = _here
SCAN_DIRS = ['app', 'migrations']
SKIP_DIRS = {'__pycache__', 'legacy code', '.venv', '.git'}
# Names that are legitimately "entry points" even with no in-repo caller.
FRAMEWORK_DECORATORS = (
'route', 'get', 'post', 'put', 'patch', 'delete', 'before_request',
'after_request', 'teardown_appcontext', 'context_processor',
'template_filter', 'errorhandler', 'cli.command', 'command',
'memoize', 'cached', 'staticmethod', 'classmethod', 'property',
'user_loader', 'login_manager',
)
LIFECYCLE_NAMES = {
'__init__', '__repr__', '__str__', '__eq__', '__hash__', '__len__',
'to_dict', 'set_password', 'check_password', 'set_quickconnect_code',
'check_quickconnect_code', 'authenticate', 'main', 'create_app',
}
# Flask auto-invoked hooks / model properties serialised by templates.
TEMPLATE_OR_HOOK_RE = re.compile(r'^(is_|has_|get_|_default|before_|after_)')
def iter_py_files():
for base in SCAN_DIRS:
root_dir = os.path.join(REPO, base)
if not os.path.isdir(root_dir):
continue
for root, dirs, files in os.walk(root_dir):
dirs[:] = [d for d in dirs if d not in SKIP_DIRS]
for f in sorted(files):
if f.endswith('.py'):
yield os.path.join(root, f)
def relpath(p):
return os.path.relpath(p, REPO)
def decorator_name(dec):
"""Best-effort dotted name for a decorator node."""
node = dec.func if isinstance(dec, ast.Call) else dec
parts = []
while isinstance(node, ast.Attribute):
parts.append(node.attr)
node = node.value
if isinstance(node, ast.Name):
parts.append(node.id)
return '.'.join(reversed(parts))
class ModuleInfo:
def __init__(self, path, tree):
self.path = path
self.rel = relpath(path)
self.tree = tree
self.imports = {} # local alias -> (module, original name)
self.module_defs = set() # top-level func/class names defined here
self.functions = [] # dicts describing each def
self.classes = []
self.calls = [] # (caller_qualname, callee_name, lineno)
def collect_module(path):
src = open(path, encoding='utf-8', errors='replace').read()
tree = ast.parse(src, filename=path)
mi = ModuleInfo(path, tree)
# ---- imports -----------------------------------------------------------
for node in ast.walk(tree):
if isinstance(node, ast.Import):
for a in node.names:
mi.imports[a.asname or a.name.split('.')[0]] = (a.name, None)
elif isinstance(node, ast.ImportFrom):
mod = node.module or ''
for a in node.names:
mi.imports[a.asname or a.name] = (mod, a.name)
# ---- top-level definitions --------------------------------------------
for node in tree.body:
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
mi.module_defs.add(node.name)
elif isinstance(node, ast.ClassDef):
mi.module_defs.add(node.name)
# ---- functions & methods ----------------------------------------------
def visit_func(node, class_name=None):
decs = [decorator_name(d) for d in node.decorator_list]
is_method = class_name is not None
# Methods are marked as framework-invoked to avoid false "unused".
entry = {
'qualname': f'{relpath(path)}::{class_name + "." if class_name else ""}{node.name}',
'name': node.name,
'file': relpath(path),
'class': class_name,
'lineno': node.lineno,
'end_lineno': getattr(node, 'end_lineno', node.lineno),
'args': [a.arg for a in node.args.args],
'decorators': decs,
'is_method': is_method,
'is_private': node.name.startswith('_') and not node.name.startswith('__'),
}
# route detection
for d in decs:
if d.endswith('route') or d in ('get', 'post', 'put', 'delete', 'patch'):
entry['route'] = True
mi.functions.append(entry)
for child in ast.walk(node):
if isinstance(child, ast.Call):
fn = child.func
name = None
if isinstance(fn, ast.Name):
name = fn.id
elif isinstance(fn, ast.Attribute):
name = fn.attr
if name:
mi.calls.append((entry['qualname'], name, child.lineno))
# nested funcs get visited separately below via walk on module
for node in tree.body:
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
visit_func(node)
elif isinstance(node, ast.ClassDef):
bases = []
for b in node.bases:
if isinstance(b, ast.Name):
bases.append(b.id)
elif isinstance(b, ast.Attribute):
bases.append(b.attr)
mi.classes.append({
'name': node.name,
'file': relpath(path),
'lineno': node.lineno,
'bases': bases,
'methods': [n.name for n in node.body
if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef))],
})
for sub in node.body:
if isinstance(sub, (ast.FunctionDef, ast.AsyncFunctionDef)):
visit_func(sub, class_name=node.name)
return mi
def main():
ap = argparse.ArgumentParser()
ap.add_argument('--json', action='store_true')
ap.add_argument('--out', default=None)
args = ap.parse_args()
modules = []
for path in iter_py_files():
try:
modules.append(collect_module(path))
except SyntaxError as e:
print(f'!! syntax error {relpath(path)}: {e}', file=sys.stderr)
# ---- global name index -------------------------------------------------
# name -> list of qualnames defining it
defined = defaultdict(list)
for mi in modules:
for f in mi.functions:
defined[f['name']].append(f['qualname'])
# ---- call graph --------------------------------------------------------
# callee name -> set of caller qualnames
callers = defaultdict(set)
for mi in modules:
for caller, callee, _ in mi.calls:
callers[callee].add(caller)
# ---- classify every function ------------------------------------------
records = []
for mi in modules:
for f in mi.functions:
q = f['qualname']
name = f['name']
decs = f['decorators']
has_framework_dec = any(
any(d.split('.')[-1] == fd or d.endswith(fd) for fd in FRAMEWORK_DECORATORS)
for d in decs
) or f.get('route')
in_repo_callers = sorted(c for c in callers.get(name, set()) if c != q)
if f.get('route'):
category = 'route'
elif has_framework_dec:
category = 'framework-hook'
elif name in LIFECYCLE_NAMES or name.startswith('__'):
category = 'dunder/lifecycle'
elif f['is_method']:
category = 'method'
elif TEMPLATE_OR_HOOK_RE.match(name):
category = 'property-like'
else:
category = 'function'
if category in ('function', 'method') and not in_repo_callers and not f['is_private']:
status = 'NO-IN-REPO-CALLER'
elif category in ('function', 'method') and not in_repo_callers and f['is_private']:
status = 'UNUSED-PRIVATE'
else:
status = 'called/hook'
records.append({
**{k: f[k] for k in ('qualname', 'name', 'file', 'class', 'lineno',
'end_lineno', 'args', 'decorators', 'is_method',
'is_private')},
'loc': f['end_lineno'] - f['lineno'] + 1,
'category': category,
'status': status,
'in_repo_callers': in_repo_callers,
'caller_count': len(in_repo_callers),
'calls': sorted({c for (a, c, _) in mi.calls if a == q}),
'duplicate_definitions': defined[name] if len(defined[name]) > 1 else [],
})
# ---- module-level import graph ----------------------------------------
import_edges = []
for mi in modules:
for alias, (mod, orig) in mi.imports.items():
if mod and (mod.startswith('app') or mod in ('app',)):
import_edges.append({'from': mi.rel, 'to': mod, 'name': orig})
# `from app.x import Y` where Y is a module
for alias, (mod, orig) in mi.imports.items():
if mod.startswith('app.') and orig is None:
import_edges.append({'from': mi.rel, 'to': mod})
result = {
'summary': {
'modules': len(modules),
'functions_total': len(records),
'files': sorted(mi.rel for mi in modules),
},
'functions': records,
'classes': [c for mi in modules for c in mi.classes],
'callers_index': {k: sorted(v) for k, v in callers.items()},
}
out = json.dumps(result, indent=2)
if args.out:
open(args.out, 'w').write(out)
print(f'wrote {args.out}')
elif args.json:
print(out)
else:
print(report(result))
def report(res):
lines = []
s = res['summary']
lines.append(f"modules: {s['modules']} functions: {s['functions_total']}")
lines.append('')
by_cat = defaultdict(list)
for f in res['functions']:
by_cat[f['category']].append(f)
for cat in sorted(by_cat):
fs = by_cat[cat]
lines.append(f'== {cat} ({len(fs)}) ==')
for f in sorted(fs, key=lambda x: (x['file'], x['lineno'])):
calls = len(f['calls'])
lines.append(f" [{f['status']:>20}] {f['file']}:{f['lineno']:<5} "
f"{f['qualname'].split('::')[1]:<45} loc={f['loc']:<4} "
f"callers={f['caller_count']:<3} calls={calls}")
lines.append('')
# suspicious: duplicate function names across files
dup = [f for f in res['functions'] if f['duplicate_definitions']]
if dup:
lines.append('== duplicate function names (same name defined in >1 place) ==')
seen = set()
for f in sorted(dup, key=lambda x: x['name']):
key = f['name']
if key in seen:
continue
seen.add(key)
lines.append(f" {key}: {f['duplicate_definitions']}")
lines.append('')
return '\n'.join(lines)
if __name__ == '__main__':
main()
+304
View File
@@ -0,0 +1,304 @@
"""Accurate sanitization report for digiserver-v2.
Corrects two things the first pass got wrong:
* blueprint -> file mapping now follows what app.py actually imports
* route usage counts url_for() AND hardcoded URL paths in templates/JS
Also classifies routes that are intentionally consumed by the external Kivy
player rather than by any template.
Run: python tools/sanitize_report.py [--markdown out.md]
"""
from __future__ import annotations
import argparse
import ast
import os
import re
from collections import defaultdict
# Walk up until we find the project root (robust to the tool living in a
# nested folder such as docs/tools/).
_here = os.path.dirname(os.path.abspath(__file__))
while _here != os.path.dirname(_here) and not os.path.isdir(os.path.join(_here, 'app')):
_here = os.path.dirname(_here)
REPO = _here
APP = os.path.join(REPO, 'app')
def read(p):
return open(p, encoding='utf-8', errors='replace').read()
def rel(p):
return os.path.relpath(p, REPO)
def walk_py(roots, skip=('__pycache__', 'legacy code')):
for root_dir in roots:
if not os.path.isdir(root_dir):
continue
for root, dirs, files in os.walk(root_dir):
dirs[:] = [d for d in dirs if d not in skip]
for f in sorted(files):
if f.endswith('.py'):
yield os.path.join(root, f)
py_files = list(walk_py([APP, os.path.join(REPO, 'migrations')]))
src = {rel(p): read(p) for p in py_files}
app_src = src.get('app/app.py', '')
# ── 1. Which blueprint modules does app.py actually import? ─────────────────
imported_bp_modules = set(re.findall(
r'from\s+(app\.blueprints\.\w+)\s+import', app_src))
registered_vars = set(re.findall(r'register_blueprint\((\w+)\)', app_src))
# ── 2. Every blueprint definition + the file that defines it ────────────────
bp_defs = [] # (var, blueprint_name, url_prefix, file, is_imported_by_app)
for path in py_files:
r = rel(path)
if not r.startswith('app/blueprints/') or os.path.basename(path).startswith('__'):
continue
s = src[r]
m = re.search(r'^(\w+_bp)\s*=\s*Blueprint\(\s*[\'"]([\w-]+)[\'"]'
r'(?:\s*,\s*[^)]*?url_prefix\s*=\s*[\'"]([^\'"]+)[\'"])?',
s, re.M | re.S)
if m:
var, bpname, prefix = m.group(1), m.group(2), m.group(3) or ''
dotted = 'app.blueprints.' + os.path.basename(path)[:-3]
bp_defs.append({
'var': var, 'name': bpname, 'prefix': prefix or '/',
'file': r, 'imported': dotted in imported_bp_modules,
'registered_var': var in registered_vars,
})
# ── 3. Collect route usage signals ─────────────────────────────────────────
url_for_refs = set()
# Literal URL strings, but only ones that look like real request paths
# (avoids matching CSS selectors, regexes, mime types, etc.).
PATH_RE = re.compile(r"['\"](/[a-z][a-zA-Z0-9_\-/<>{}.:]*)['\"]")
hardcoded_paths = set()
consider_files = []
for root, dirs, files in os.walk(os.path.join(APP, 'templates')):
for f in files:
if f.endswith(('.html', '.js')):
consider_files.append(os.path.join(root, f))
consider_files += [os.path.join(REPO, r) for r in src
if r.endswith(('.html', '.js'))]
for fp in consider_files:
if not os.path.isfile(fp):
continue
s = read(fp)
url_for_refs |= set(re.findall(r"url_for\(\s*['\"]([\w\.]+)['\"]", s))
hardcoded_paths |= set(PATH_RE.findall(s))
for r, s in src.items():
if r.endswith('.py'):
url_for_refs |= set(re.findall(r"url_for\(\s*['\"]([\w\.]+)['\"]", s))
hardcoded_paths |= set(PATH_RE.findall(s))
# ── 4. Enumerate routes ────────────────────────────────────────────────────
API_PREFIX = '/api'
routes = []
for path in py_files:
r = rel(path)
if not r.startswith('app/blueprints/'):
continue
s = src[r]
bp_name = None
file_prefix = ''
for d in bp_defs:
if d['file'] == r:
bp_name = d['name']
file_prefix = d['prefix']
try:
tree = ast.parse(s)
except SyntaxError:
continue
for node in tree.body:
if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
continue
for dec in node.decorator_list:
if not (isinstance(dec, ast.Call) and getattr(dec.func, 'attr', '') == 'route'):
continue
rule = dec.args[0].value if dec.args and isinstance(dec.args[0], ast.Constant) else '?'
methods = 'GET'
for kw in dec.keywords:
if kw.arg == 'methods' and isinstance(kw.value, ast.List):
methods = ','.join(e.value for e in kw.value.elts
if isinstance(e, ast.Constant))
full = file_prefix.rstrip('/') + rule if rule != '/' else file_prefix.rstrip('/') + '/'
endpoint = f'{bp_name}.{node.name}'
routes.append({
'endpoint': endpoint, 'func': node.name, 'rule': rule,
'full': full, 'methods': methods, 'file': r,
'lineno': node.lineno,
'url_for_ref': endpoint in url_for_refs,
'hardcoded_ref': full in hardcoded_paths or rule in hardcoded_paths,
'is_api': (file_prefix or '').startswith(API_PREFIX),
})
# ── 5. Broken model attribute references ──────────────────────────────────
model_attrs = defaultdict(set)
for path in py_files:
r = rel(path)
if not r.startswith('app/models/') or os.path.basename(path).startswith('__'):
continue
try:
tree = ast.parse(src[r])
except SyntaxError:
continue
for node in tree.body:
if isinstance(node, ast.ClassDef):
for sub in ast.walk(node):
if isinstance(sub, ast.Assign):
for t in sub.targets:
if isinstance(t, ast.Name):
model_attrs[node.name].add(t.id)
elif isinstance(sub, (ast.FunctionDef, ast.AsyncFunctionDef)):
model_attrs[node.name].add(sub.name)
VIA = {'via', 'e', 'exc', 'error', 'err', 'a', 'b', 'flask_', 'db_', 'fh', 'fd'}
# ── report ────────────────────────────────────────────────────────────────
lines = []
def out(s=''):
lines.append(s)
out('# DigiServer v2 — Code Sanitization Report')
out()
out(f'Analysed **{len(py_files)} Python files** under `app/` and `migrations/`.')
out()
out('## 1. Dead modules (never imported)')
out()
out('| File | Status |')
out('|---|---|')
dead_files = []
for d in bp_defs:
if not d['imported']:
dead_files.append(d['file'])
out(f"| `{d['file']}` | **DEAD** — defines `{d['var']}` but `app.py` imports another module |")
for path in py_files:
r = rel(path)
base = os.path.basename(path)[:-3]
if not r.startswith('app/') or r.startswith('app/blueprints/'):
continue
if base == '__init__':
continue
dotted = r[:-3].replace('/', '.')
referenced = any(dotted in s or f'.{base} import' in s
for k, s in src.items() if k != r)
if not referenced:
dead_files.append(r)
out(f'| `{r}` | **DEAD** — never imported by any module |')
if not dead_files:
out('| (none) | |')
out()
out('## 2. Blueprint registration')
out()
out('| Blueprint var | Module | Prefix | Registered |')
out('|---|---|---|---|')
for d in sorted(bp_defs, key=lambda x: x['name']):
mark = 'yes' if (d['imported'] and d['registered_var']) else '**NO — DEAD**'
out(f"| `{d['var']}` | `{d['file']}` | `{d['prefix']}` | {mark} |")
out()
used = [r for r in routes if r['url_for_ref'] or r['hardcoded_ref']]
api_routes = [r for r in routes if r['is_api']]
unused_non_api = [r for r in routes if not r['url_for_ref'] and not r['hardcoded_ref']
and not r['is_api']]
out('## 3. Routes')
out()
out(f'- Total routes: **{len(routes)}**')
out(f'- Referenced by a template/JS (`url_for` or hardcoded path): **{len(used)}**')
out(f'- API routes (`/api/*`, consumed by the external Kivy player, not templates): **{len(api_routes)}**')
out(f'- Non-API routes with **no template reference**: **{len(unused_non_api)}**')
out()
out('### 3a. Non-API routes with no template reference (candidates)')
out()
out('| Endpoint | Methods | Path | File:line |')
out('|---|---|---|---|')
for r in sorted(unused_non_api, key=lambda x: (x['file'], x['lineno'])):
out(f"| `{r['endpoint']}` | {r['methods']} | `{r['full']}` | `{r['file']}:{r['lineno']}` |")
out()
out('## 4. Broken attribute references (would raise at runtime)')
out()
out('| Location | Reference | Problem |')
out('|---|---|---|')
# Locals that are known to hold a specific model instance.
VAR_TO_MODEL = {
'player': 'Player', 'content': 'Content', 'playlist': 'Playlist',
'assigned_playlist': 'Playlist', 'feedback': 'PlayerFeedback',
'latest_feedback': 'PlayerFeedback', 'edit_record': 'PlayerEdit',
'admin': 'User', 'user': 'User', 'log': 'ServerLog',
'group': 'Group', 'new_user': 'PlayerUser', 'existing_user': 'PlayerUser',
}
# Python/library attributes that would otherwise be false positives.
SAFE = {
'query', 'get', 'filter_by', 'first', 'all', 'count', 'id', 'session', 'add',
'commit', 'rollback', 'delete', 'get_or_404', 'order_by', 'isoformat', 'route',
'methods', 'get_json', 'args', 'files', 'form', 'json', 'headers', 'values',
'remote_addr', 'host_url', 'script_root', 'scheme', 'host', 'root_path',
'config', 'utcnow', 'now', 'total_seconds', 'name', 'value', 'keys', 'items',
'groups', 'contents', 'players', 'append', 'lower', 'upper', 'strip', 'split',
'join', 'replace', 'encode', 'decode', 'check_password', 'check_quickconnect_code',
'set_password', 'set_quickconnect_code', 'authenticate', 'update_status',
'to_dict', 'is_online', 'is_admin', 'is_active', 'file_size_mb', 'group_count',
'player_count', 'content_count', 'original_display_name', 'original_media_path',
'current_media_path', 'get_content_ordered', 'version', 'filename', 'content_type',
'duration', 'url', 'description', 'file_size', 'uploaded_at', 'hostname',
'location', 'auth_code', 'orientation', 'status', 'playlist_id', 'last_seen',
'created_at', 'updated_at', 'deployment_status', 'last_deployment_at',
'last_deployment_status', 'last_deployment_message', 'original_filename',
'time_of_modification', 'metadata_path', 'edited_file_path', 'new_name',
'original_name', 'username', 'role', 'password', 'level', 'message', 'user_code',
'user_name', 'email', 'func', 'rules', 'player_id', 'content_id', 'error',
'show', 'seek', 'load', 'play', 'pause', 'stop', 'text', 'bind', 'add_widget',
'clear_widgets', 'current', 'parent', 'children', 'ids', 'size', 'pos', 'opacity',
'source', 'state', 'duration', 'position', 'muted', 'audio', 'describe',
}
broken = []
for r, s in src.items():
if not (r.startswith('app/blueprints/') or r.startswith('app/utils/')):
continue
try:
tree = ast.parse(s)
except SyntaxError:
continue
for node in ast.walk(tree):
if not (isinstance(node, ast.Attribute) and isinstance(node.value, ast.Name)):
continue
var, attr = node.value.id, node.attr
model = VAR_TO_MODEL.get(var)
if not model or model not in model_attrs:
continue
if attr in model_attrs[model] or attr in SAFE:
continue
broken.append((r, node.lineno, f'{var}.{attr}', f'`{model}` has no `{attr}`'))
for r, ln, ref, prob in broken:
out(f'| `{r}:{ln}` | `{ref}` | {prob} |')
if not broken:
out('| (none) | | |')
out()
out('## 5. How to use this report')
out()
out('Each section lists deletion candidates. Reply with the section/symbol names')
out('you want removed and they will be deleted (the snapshot in `legacy code/`')
out('preserves the originals).')
text = '\n'.join(lines)
ap = argparse.ArgumentParser()
ap.add_argument('--markdown', default=None)
args = ap.parse_args()
if args.markdown:
open(args.markdown, 'w').write(text)
print(f'wrote {args.markdown}')
else:
print(text)
+65
View File
@@ -0,0 +1,65 @@
"""Detect orphan templates (never rendered) and orphan static assets."""
from __future__ import annotations
import os
import re
# Walk up until we find the project root (robust to the tool living in a
# nested folder such as docs/tools/).
_here = os.path.dirname(os.path.abspath(__file__))
while _here != os.path.dirname(_here) and not os.path.isdir(os.path.join(_here, 'app')):
_here = os.path.dirname(_here)
REPO = _here
APP = os.path.join(REPO, 'app')
TPL = os.path.join(APP, 'templates')
def read(p):
return open(p, encoding='utf-8', errors='replace').read()
# gather all render_template('x.html') calls
rendered = set()
for root, dirs, files in os.walk(APP):
dirs[:] = [d for d in dirs if d not in ('__pycache__', 'templates')]
for f in files:
if f.endswith('.py'):
s = read(os.path.join(root, f))
rendered |= set(re.findall(r"render_template\(\s*['\"]([^'\"]+)['\"]", s))
# render_template with variable name - note it
rendered |= set(re.findall(r"render_template\(\s*([a-z_]+)", s)) - {'f'}
# gather extends/include/import references
refs = set()
for root, dirs, files in os.walk(TPL):
for f in files:
if f.endswith('.html'):
s = read(os.path.join(root, f))
for pat in (r"{%\s*extends\s*['\"]([^'\"]+)['\"]",
r"{%\s*include\s*['\"]([^'\"]+)['\"]",
r"{%\s*import\s*['\"]([^'\"]+)['\"]",
r"render_template\(\s*['\"]([^'\"]+)['\"]"):
refs |= set(re.findall(pat, s))
all_tpl = []
for root, dirs, files in os.walk(TPL):
for f in files:
if f.endswith('.html'):
all_tpl.append(os.path.relpath(os.path.join(root, f), TPL))
referenced = rendered | refs
orphans = []
for t in sorted(all_tpl):
base = os.path.basename(t)
if t in referenced or base in referenced:
continue
# macro/library files are referenced dynamically sometimes
orphans.append(t)
print(f'templates total : {len(all_tpl)}')
print(f'referenced : {len([t for t in all_tpl if t not in orphans])}')
print(f'ORPHAN templates : {len(orphans)}')
print()
for o in orphans:
lines = len(read(os.path.join(TPL, o)).splitlines())
print(f'{o:<58} {lines} lines')
+108
View File
@@ -0,0 +1,108 @@
"""Post-sanitization verification smoke test.
Boots the app on a fresh DB, registers a player/playlist/content, and exercises
every API endpoint plus key UI routes. Fails loudly on any 5xx.
"""
import os
import tempfile
tmpdir = tempfile.mkdtemp()
db_path = os.path.join(tmpdir, 'smoke.db')
os.environ['DATABASE_URL'] = f'sqlite:///{db_path}'
from app.app import create_app
from app.extensions import db, bcrypt
from app.models import Player, Playlist, Content, User, PlayerUser, HTTPSConfig
app = create_app('testing')
app.config['SQLALCHEMY_DATABASE_URI'] = f'sqlite:///{db_path}'
app.config['CACHE_TYPE'] = 'simple'
with app.app_context():
db.create_all()
# prove Group/group_content are gone from metadata
tables = sorted(db.metadata.tables.keys())
print('registered tables:', tables)
assert 'group' not in tables, 'group table still registered!'
assert 'group_content' not in tables, 'group_content still registered!'
admin = User(username='smoke',
password=bcrypt.generate_password_hash('smokepw').decode('utf-8'),
role='admin')
pl = Playlist(name='Smoke PL')
db.session.add_all([admin, pl])
db.session.flush()
c = Content(filename='s.png', content_type='image', duration=5)
db.session.add(c)
db.session.flush()
pl.contents.append(c)
p = Player(name='Smoke', hostname='smoke-host', auth_code='smoke-code')
p.set_password('pw')
p.set_quickconnect_code('qc')
p.playlist_id = pl.id
db.session.add(p)
db.session.commit()
pid = p.id
code = p.auth_code
auth = {'Authorization': f'Bearer {code}'}
with app.test_client() as cl:
# public api
for url in ['/api/health', '/api/system-info', '/api/content', '/api/logs',
f'/api/player-status/{pid}',
'/api/playlists?hostname=smoke-host&quickconnect_code=qc']:
r = cl.get(url)
print(f'{r.status_code} GET {url}')
assert r.status_code < 500, f'{url} -> {r.status_code}'
# system-info must no longer expose a groups key
if url == '/api/system-info':
body = r.get_json()
assert 'groups' not in body, f'groups key still present: {body}'
print(' system-info keys:', sorted(body.keys()))
# authed api
for url in [f'/api/playlists/{pid}', f'/api/playlist-version/{pid}']:
r = cl.get(url, headers=auth)
print(f'{r.status_code} GET {url} (bearer)')
assert r.status_code < 500
r = cl.post('/api/auth/verify', json={'auth_code': code})
print(f'{r.status_code} POST /api/auth/verify')
assert r.status_code < 500
r = cl.post('/api/player-feedback', json={
'hostname': 'smoke-host', 'quickconnect_code': 'qc',
'status': 'playing', 'message': 'ok'})
print(f'{r.status_code} POST /api/player-feedback')
assert r.status_code < 500
# content listing must no longer contain group_count
body = cl.get('/api/content').get_json()
assert body['content'], 'expected seeded content'
assert 'group_count' not in body['content'][0], body['content'][0]
print(' /api/content item keys:', sorted(body['content'][0].keys()))
# login then hit UI pages
r = cl.post('/login', data={'username': 'smoke', 'password': 'smokepw'},
follow_redirects=True)
print(f'{r.status_code} POST /login')
for url in ['/', '/content/', '/content/media-library', '/players/',
'/admin/', '/admin/users', '/admin/system/info']:
r = cl.get(url, follow_redirects=True)
print(f'{r.status_code} GET {url}')
assert r.status_code < 500, f'{url} -> {r.status_code}'
# import checks
import app.utils as u
print('utils exports ok:', hasattr(u, 'get_player_status_info'))
assert not hasattr(u, 'get_group_statistics')
assert not hasattr(u, 'assign_player_to_group')
import importlib
for mod in ['app.blueprints.content', 'app.blueprints.api', 'app.utils.group_player_management',
'app.blueprints.players']:
importlib.import_module(mod)
print(f'imported {mod}')
print('\n🎉 SMOKE TEST PASSED')
+97
View File
@@ -0,0 +1,97 @@
"""Drive the real player build through the HTTP UI and poll its progress.
Logs in as admin over HTTPS, POSTs the build form exactly like the browser, then
polls /admin/build-player/status until it finishes — mirroring what a real user
sees, including the non-blocking behaviour.
Optional arg: branch to build (default 'main').
"""
import http.cookiejar
import json
import re
import ssl
import sys
import time
import urllib.parse
import urllib.request
BASE = 'https://192.168.0.152'
BRANCH = sys.argv[1] if len(sys.argv) > 1 else 'main'
pw = open('.deployment-credentials').read().split('admin password: ')[1].strip()
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
op = urllib.request.build_opener(
urllib.request.HTTPCookieProcessor(http.cookiejar.CookieJar()),
urllib.request.HTTPSHandler(context=ctx),
)
def get(path):
return op.open(BASE + path, timeout=60).read().decode()
print('1) logging in…')
html = get('/login')
tok = re.search(r'name="csrf_token"[^>]*value="([^"]+)"', html)
data = urllib.parse.urlencode({
'username': 'admin', 'password': pw,
'csrf_token': tok.group(1) if tok else '',
}).encode()
r = op.open(urllib.request.Request(BASE + '/login', data=data, method='POST'), timeout=60)
print(' logged in ->', r.geturl())
print(f'2) posting build form (branch={BRANCH})…')
html = get('/admin/build-player')
tok = re.search(r'name="csrf_token"[^>]*value="([^"]+)"', html)
form = urllib.parse.urlencode({
'action': 'build_and_config',
'repo_url': 'https://gitea.moto-adv.com/ske087/Kiwy-Signage.git',
'branch': BRANCH,
'server_ip': '192.168.0.152',
'port': '443',
'use_https': 'on',
'orientation': 'Landscape',
'max_resolution': '1920x1080',
'csrf_token': tok.group(1) if tok else '',
}).encode()
t0 = time.time()
r = op.open(urllib.request.Request(BASE + '/admin/build-player', data=form, method='POST'),
timeout=60)
post_elapsed = time.time() - t0
print(f' POST returned in {post_elapsed:.1f}s -> {r.status} {r.geturl()}')
if post_elapsed > 30:
print(' ⚠ POST blocked for a long time — the build is NOT async!')
print('3) polling status…')
last = None
deadline = time.time() + 420
while time.time() < deadline:
try:
raw = get('/admin/build-player/status')
s = json.loads(raw)
except Exception as e: # noqa: BLE001
time.sleep(2)
continue
cur = (s.get('state'), s.get('step'), s.get('message'))
if cur != last:
print(f" [{time.time() - t0:6.1f}s] state={s.get('state'):8} "
f"step={s.get('step') or '-':28} version={s.get('version')}")
if s.get('message'):
print(f" msg: {s['message'][:110]}")
last = cur
if s.get('state') in ('success', 'error'):
print()
print('RESULT:', s.get('state'))
print(' version :', s.get('version'))
print(' message :', (s.get('message') or '')[:400])
sys.exit(0 if s.get('state') == 'success' else 1)
time.sleep(2)
print('timed out waiting for the build')
sys.exit(1)
+179
View File
@@ -0,0 +1,179 @@
#!/bin/bash
# End-to-end runtime test of the HTTP/HTTPS + fallback behaviour.
#
# Spins up a stub backend + Caddy on a throwaway Docker network and proves:
# 1. HTTPS disabled -> http://<ip>:PORT answers on plain HTTP
# 2. HTTPS enabled -> https://<ip>:PORT answers over TLS (internal CA)
# 3. HTTPS enabled -> http://<ip>:PORT STILL answers (fallback)
# 4. Both the IP and a hostname resolve to the app
#
# Uses a stub backend so no 1.17 GB app image build is needed; the reverse-proxy
# behaviour under test is entirely Caddy's.
set -u
NET=e2e-caddy-net
BACKEND=e2e-backend
CADDY=e2e-caddy
HTTP_PORT=18080
HTTPS_PORT=18443
IP=127.0.0.1
cleanup() {
docker rm -f "$CADDY" "$BACKEND" >/dev/null 2>&1 || true
docker network rm "$NET" >/dev/null 2>&1 || true
rm -f /tmp/e2e_Caddyfile
}
trap cleanup EXIT
cleanup
docker network create "$NET" >/dev/null
# Stub "digiserver-app" serving a recognisable body on :5000
docker run -d --name "$BACKEND" --network "$NET" --network-alias digiserver-app \
python:3.13-slim \
python -c "from http.server import BaseHTTPRequestHandler,HTTPServer
class H(BaseHTTPRequestHandler):
def do_GET(self):
self.send_response(200); self.send_header('Content-Type','text/plain'); self.end_headers()
self.wfile.write(b'BACKEND-OK')
def log_message(self,*a): pass
HTTPServer(('0.0.0.0',5000),H).serve_forever()" >/dev/null
echo "waiting for stub backend..."
for i in $(seq 1 20); do
docker exec "$BACKEND" python -c "
import urllib.request,sys
try:
urllib.request.urlopen('http://localhost:5000/',timeout=1); sys.exit(0)
except Exception: sys.exit(1)" 2>/dev/null && break
sleep 1
done
echo "stub backend ready"
echo
start_caddy() { # $1 = caddyfile content
printf '%s' "$1" > /tmp/e2e_Caddyfile
docker rm -f "$CADDY" >/dev/null 2>&1 || true
docker run -d --name "$CADDY" --network "$NET" \
-p "${HTTP_PORT}:80" -p "${HTTPS_PORT}:443" \
-v /tmp/e2e_Caddyfile:/etc/caddy/Caddyfile:ro \
caddy:2-alpine >/dev/null
# wait for the admin API to accept connections
for i in $(seq 1 25); do
docker exec "$CADDY" wget -q -O- http://localhost:2019/config/ >/dev/null 2>&1 && return 0
sleep 1
done
return 1
}
result() { # $1 label, $2 expected substring, $3 actual body
if printf '%s' "$3" | grep -q "$2"; then
echo " PASS $1"
return 0
fi
echo " FAIL $1 (got: $(printf '%s' "$3" | head -c 80))"
return 1
}
FAILED=0
# ── Case 1: HTTPS disabled -> plain HTTP only ────────────────────────────────
echo "CASE 1: HTTPS disabled -> plain HTTP on port $HTTP_PORT"
start_caddy '{
admin 0.0.0.0:2019
}
:80 {
reverse_proxy digiserver-app:5000
}
' || echo " (caddy admin not ready; continuing)"
BODY=$(curl -sS -m 5 "http://${IP}:${HTTP_PORT}/" 2>&1)
result "http://IP answers" "BACKEND-OK" "$BODY" || FAILED=1
BODY=$(curl -sS -m 5 -H "Host: digiserver" "http://${IP}:${HTTP_PORT}/" 2>&1)
result "http with Host: digiserver answers (catch-all)" "BACKEND-OK" "$BODY" || FAILED=1
echo
# ── Case 2/3: HTTPS on, internal CA, with HTTP fallback ──────────────────────
echo "CASE 2+3: HTTPS on (internal CA) + HTTP fallback"
# Generated by CaddyConfigGenerator for ip=127.0.0.1, hostname=digiserver.
# `default_sni` is REQUIRED: browsers send no SNI when the URL is an IP, so
# without it Caddy matches no certificate and aborts with
# "no certificate available for '<container-ip>'".
start_caddy "{
admin 0.0.0.0:2019
email admin@example.com
default_sni ${IP}
}
:80 {
reverse_proxy digiserver-app:5000
}
http://${IP} {
reverse_proxy digiserver-app:5000
}
http://digiserver {
reverse_proxy digiserver-app:5000
}
https://${IP} {
tls internal
reverse_proxy digiserver-app:5000
}
https://digiserver {
tls internal
reverse_proxy digiserver-app:5000
}
" || echo " (caddy admin not ready; continuing)"
echo " (waiting for internal CA issuance)"
for i in $(seq 1 15); do
OUT=$(curl -sS -k -m 4 "https://${IP}:${HTTPS_PORT}/" 2>&1)
printf '%s' "$OUT" | grep -q "BACKEND-OK" && break
sleep 1
done
result "https://IP answers over TLS (SNI-less)" "BACKEND-OK" "$OUT" || FAILED=1
OUT2=$(curl -sS -k -m 6 --resolve "digiserver:${HTTPS_PORT}:${IP}" \
"https://digiserver:${HTTPS_PORT}/" 2>&1)
result "https://hostname answers over TLS (with SNI)" "BACKEND-OK" "$OUT2" || FAILED=1
BODY=$(curl -sS -m 5 "http://${IP}:${HTTP_PORT}/" 2>&1)
result "http://IP STILL answers (fallback)" "BACKEND-OK" "$BODY" || FAILED=1
BODY=$(curl -sS -m 5 -H "Host: digiserver" "http://${IP}:${HTTP_PORT}/" 2>&1)
result "http:// with Host: digiserver answers" "BACKEND-OK" "$BODY" || FAILED=1
HANDSHAKE_ERRORS=$(docker logs "$CADDY" 2>&1 | grep -ci "handshake error" || true)
if [ "$HANDSHAKE_ERRORS" -eq 0 ]; then
echo " PASS no TLS handshake errors logged"
else
echo " FAIL $HANDSHAKE_ERRORS TLS handshake error(s) logged"
FAILED=1
fi
# Certificate must come from Caddy's local CA
ISSUER=$(echo | openssl s_client -connect "${IP}:${HTTPS_PORT}" -servername localhost 2>/dev/null \
| openssl x509 -noout -issuer 2>/dev/null)
if printf '%s' "$ISSUER" | grep -qi "local\|caddy"; then
echo " PASS certificate issued by the internal CA"
echo " $ISSUER"
else
echo " WARN unexpected issuer: ${ISSUER:-none}"
fi
echo
docker logs "$CADDY" 2>&1 | grep -iE "error|cannot|fail" | head -5 || true
echo
if [ "$FAILED" -eq 0 ]; then
echo "ALL RUNTIME CHECKS PASSED"
else
echo "SOME RUNTIME CHECKS FAILED"
fi
exit "$FAILED"
+232
View File
@@ -0,0 +1,232 @@
"""Verify the env-driven HTTPS bootstrap in https_manager.py.
Covers the behaviours the user asked for:
1. Both HOSTNAME_INTERNAL + HOST_IP set -> HTTPS configured at startup
2. Either missing -> NO-OP, HTTP fallback stays
3. Re-running with the same env -> idempotent
4. HTTPS_HTTP_FALLBACK=false -> redirect to the published port
5. Admin UI change after bootstrap -> overrides the env value
Run: PYTHONPATH=$(pwd) ./.venv/bin/python docs/tools/test_https_bootstrap.py
"""
import importlib.util
import os
import shutil
import tempfile
# DATABASE_URL must be set BEFORE importing the app: ProductionConfig evaluates
# it at class-definition (import) time.
TMPDIR = tempfile.mkdtemp()
DB_PATH = os.path.join(TMPDIR, 'bootstrap.db')
os.environ['DATABASE_URL'] = f'sqlite:///{DB_PATH}'
from app.app import create_app # noqa: E402
from app.extensions import db # noqa: E402
from app.models.https_config import HTTPSConfig # noqa: E402
REPO = '/home/scheianu/digiserver-v2'
TARGET = os.path.join(TMPDIR, 'Caddyfile')
PLACEHOLDER = ':80 {\n respond "placeholder"\n}\n'
app = create_app('production')
def reset_db():
"""Drop and recreate the schema for a clean case."""
with app.app_context():
db.drop_all()
db.create_all()
def load_manager():
"""Import https_manager with file writes/reloads redirected to TARGET."""
spec = importlib.util.spec_from_file_location(
'hm_boot', os.path.join(REPO, 'https_manager.py'))
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod)
mod.CaddyConfigGenerator.write_caddyfile = staticmethod(
lambda content, path=TARGET: (open(TARGET, 'w').write(content), True)[1])
mod.CaddyConfigGenerator.reload_caddy = staticmethod(lambda: True)
return mod
def reset_env():
for k in ('HOSTNAME_INTERNAL', 'HOST_IP', 'DOMAIN', 'SSL_EMAIL',
'HTTPS_PORT', 'HTTPS_HTTP_FALLBACK', 'HTTPS_VERIFY'):
os.environ.pop(k, None)
# No live Caddy in these unit tests, so skip the post-reload TLS probe.
# The probe + automatic fallback are covered by
# docs/tools/test_https_fallback.py.
os.environ['HTTPS_VERIFY'] = 'false'
def blocks():
return [ln.strip() for ln in open(TARGET).read().splitlines()
if ln.strip().startswith(('http://', 'https://', ':80'))]
def config():
with app.app_context():
return HTTPSConfig.get_config()
print('=' * 68)
print('CASE 1 — both vars set -> HTTPS configured at startup')
print('=' * 68)
reset_db(); reset_env()
open(TARGET, 'w').write(PLACEHOLDER)
os.environ.update(HOSTNAME_INTERNAL='digiserver', HOST_IP='192.168.0.152',
SSL_EMAIL='admin@example.com')
hm = load_manager()
rc = hm.bootstrap_from_env(app)
c = config()
print('exit code:', rc, '| https_enabled:', c.https_enabled, '| domain:', repr(c.domain))
print('blocks:', blocks())
assert rc == 0
assert c.https_enabled is True
assert not c.domain, 'empty domain must select the internal CA'
assert 'tls internal' in open(TARGET).read()
assert 'http://192.168.0.152' in open(TARGET).read()
print('PASS: internal CA + HTTP fallback generated\n')
print('=' * 68)
print('CASE 2 — HOST_IP missing -> no-op, HTTP fallback remains')
print('=' * 68)
reset_db(); reset_env()
open(TARGET, 'w').write(PLACEHOLDER)
os.environ['HOSTNAME_INTERNAL'] = 'digiserver' # HOST_IP absent
hm = load_manager()
rc = hm.bootstrap_from_env(app)
c = config()
print('exit code:', rc, '| config:', 'none' if c is None else c.https_enabled)
assert rc == 0, 'must not fail container startup'
assert c is None or c.https_enabled is False, 'HTTPS must stay off'
assert open(TARGET).read() == PLACEHOLDER, 'Caddyfile must be untouched'
print('PASS: stayed on plain-HTTP fallback, Caddyfile untouched\n')
print('=' * 68)
print('CASE 3 — HOSTNAME_INTERNAL missing -> no-op')
print('=' * 68)
reset_db(); reset_env()
os.environ['HOST_IP'] = '192.168.0.152'
hm = load_manager()
rc = hm.bootstrap_from_env(app)
c = config()
print('exit code:', rc, '| config:', 'none' if c is None else c.https_enabled)
assert c is None or c.https_enabled is False
print('PASS: no-op when hostname is missing\n')
print('=' * 68)
print('CASE 3b — neither var set (default fresh deploy) -> no-op')
print('=' * 68)
reset_db(); reset_env()
hm = load_manager()
rc = hm.bootstrap_from_env(app)
assert config() is None
print('exit code:', rc, '| https_config row: none')
print('PASS: untouched config, HTTP fallback only\n')
print('=' * 68)
print('CASE 4 — idempotency (run twice)')
print('=' * 68)
reset_db(); reset_env()
open(TARGET, 'w').write(PLACEHOLDER)
os.environ.update(HOSTNAME_INTERNAL='digiserver', HOST_IP='192.168.0.152')
hm = load_manager()
hm.bootstrap_from_env(app)
first = open(TARGET).read()
hm.bootstrap_from_env(app)
second = open(TARGET).read()
with app.app_context():
n = HTTPSConfig.query.count()
print('https_config rows:', n, '| Caddyfile identical:', first == second)
assert n == 1, f'expected one config row, got {n}'
assert first == second, 'Caddyfile changed on re-run'
print('PASS: idempotent\n')
print('=' * 68)
print('CASE 5 — HTTPS_HTTP_FALLBACK=false -> redirect to the HTTPS URL')
print('=' * 68)
reset_db(); reset_env()
open(TARGET, 'w').write(PLACEHOLDER)
# HTTPS_PORT stays at its 443 default, so the redirect target must not carry
# an explicit port (https://host, not https://host:443).
os.environ.update(HOSTNAME_INTERNAL='digiserver', HOST_IP='192.168.0.152',
HTTPS_HTTP_FALLBACK='false')
hm = load_manager()
hm.bootstrap_from_env(app)
text = open(TARGET).read()
print('blocks:', blocks())
assert 'redir https://192.168.0.152{uri} 301' in text, \
'redirect should omit :443 when HTTPS_PORT is 443'
print('PASS: redirect targets https://<ip> with no redundant port\n')
print('=' * 68)
print('CASE 5b — non-standard HTTPS_PORT -> redirect includes the port')
print('=' * 68)
reset_db(); reset_env()
open(TARGET, 'w').write(PLACEHOLDER)
os.environ.update(HOSTNAME_INTERNAL='digiserver', HOST_IP='192.168.0.152',
HTTPS_PORT='8443', HTTPS_HTTP_FALLBACK='false')
hm = load_manager()
hm.bootstrap_from_env(app)
text = open(TARGET).read()
print('blocks:', blocks())
assert 'redir https://192.168.0.152:8443{uri} 301' in text, \
'redirect must include a non-standard port'
print('PASS: redirect includes :8443\n')
print('=' * 68)
print('CASE 6 — admin change BEFORE restart is NOT clobbered by the env')
print('=' * 68)
reset_db(); reset_env()
open(TARGET, 'w').write(PLACEHOLDER)
os.environ.update(HOSTNAME_INTERNAL='digiserver', HOST_IP='192.168.0.152')
hm = load_manager()
hm.bootstrap_from_env(app) # startup bootstrap (env owns it)
print('after bootstrap IP :', config().ip_address)
# Simulate an admin changing the IP in the UI (updated_by = the username).
hm._apply(app, https_enabled=True, hostname='digiserver', domain='',
email=None, ip_address='10.0.0.99', port=443, http_fallback=True,
verify=False)
with app.app_context():
c = HTTPSConfig.get_config()
c.updated_by = 'admin' # as admin.py would record it
from app.extensions import db as _db
_db.session.commit()
print('after admin IP :', config().ip_address)
# Next container start re-runs the bootstrap with the SAME env.
rc = hm.bootstrap_from_env(app)
print('bootstrap exit code:', rc)
print('IP after restart :', config().ip_address)
assert rc == 0
assert config().ip_address == '10.0.0.99', \
'admin setting must survive an env bootstrap on restart'
# The admin-set IP must be served (the hostname from before is still served too,
# because the admin only changed the IP field).
assert 'https://10.0.0.99 {' in open(TARGET).read(), \
'admin IP should be present in the Caddyfile'
assert 'https://192.168.0.152 {' not in open(TARGET).read(), \
'the old env IP must no longer be served'
print('PASS: admin-owned config is preserved across restarts\n')
print('=' * 68)
print('CASE 7 — env still owns config -> env change IS applied on restart')
print('=' * 68)
reset_db(); reset_env()
open(TARGET, 'w').write(PLACEHOLDER)
os.environ.update(HOSTNAME_INTERNAL='digiserver', HOST_IP='192.168.0.152')
hm = load_manager()
hm.bootstrap_from_env(app)
print('first start IP :', config().ip_address)
os.environ['HOST_IP'] = '10.20.30.40' # operator edits .env
hm.bootstrap_from_env(app)
print('second start IP:', config().ip_address)
assert config().ip_address == '10.20.30.40', 'env change should apply'
assert 'https://10.20.30.40' in open(TARGET).read()
print('PASS: env changes still take effect while env owns the config\n')
shutil.rmtree(TMPDIR, ignore_errors=True)
print('ALL BOOTSTRAP CASES PASSED')
+119
View File
@@ -0,0 +1,119 @@
"""Verify the HTTPS-verify + automatic HTTP fallback in https_manager._apply.
This is the safety net the user asked for: if HTTPS is enabled but does not
actually work, the server must not be left unreachable — it falls back to HTTP.
The TLS probe is monkey-patched so no live server is required.
Run: PYTHONPATH=$(pwd) ./.venv/bin/python docs/tools/test_https_fallback.py
"""
import importlib.util
import os
import shutil
import tempfile
TMPDIR = tempfile.mkdtemp()
os.environ['DATABASE_URL'] = f'sqlite:///{TMPDIR}/fb.db'
from app.app import create_app # noqa: E402
from app.extensions import db # noqa: E402
from app.models.https_config import HTTPSConfig # noqa: E402
REPO = '/home/scheianu/digiserver-v2'
TARGET = os.path.join(TMPDIR, 'Caddyfile')
app = create_app('production')
with app.app_context():
db.create_all()
def load_manager(verify_result):
"""Import https_manager with I/O redirected and the probe stubbed."""
spec = importlib.util.spec_from_file_location(
'hm_fb', os.path.join(REPO, 'https_manager.py'))
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod)
mod.CaddyConfigGenerator.write_caddyfile = staticmethod(
lambda content, path=TARGET: (open(TARGET, 'w').write(content), True)[1])
mod.CaddyConfigGenerator.reload_caddy = staticmethod(lambda: True)
mod.verify_https = lambda *a, **k: verify_result
return mod
def blocks():
return [ln.strip() for ln in open(TARGET).read().splitlines()
if ln.strip().startswith(('http://', 'https://', ':80'))]
def config():
with app.app_context():
return HTTPSConfig.get_config()
print('=' * 68)
print('CASE A — HTTPS verifies OK -> stays enabled')
print('=' * 68)
open(TARGET, 'w').write(':80 {\n respond "x"\n}\n')
hm = load_manager((True, 'HTTP 200 from https://192.168.0.152:443/api/health'))
rc = hm._apply(app, https_enabled=True, hostname='digiserver', domain='',
email=None, ip_address='192.168.0.152', port=443,
http_fallback=True, verify=True)
c = config()
print('exit code:', rc, '| https_enabled:', c.https_enabled)
print('blocks:', blocks())
assert rc == 0
assert c.https_enabled is True
assert 'tls internal' in open(TARGET).read()
print('PASS: HTTPS left enabled\n')
print('=' * 68)
print('CASE B — HTTPS probe FAILS -> automatic HTTP fallback')
print('=' * 68)
open(TARGET, 'w').write(':80 {\n respond "x"\n}\n')
hm = load_manager((False, 'ConnectionRefusedError: refused'))
rc = hm._apply(app, https_enabled=True, hostname='digiserver', domain='',
email=None, ip_address='192.168.0.152', port=443,
http_fallback=True, verify=True)
c = config()
text = open(TARGET).read()
print('exit code:', rc, '| https_enabled:', c.https_enabled)
print('blocks:', blocks())
assert rc == 1, f'expected exit 1 signalling the fallback, got {rc}'
assert c.https_enabled is False, 'must revert to HTTP-only'
assert 'tls internal' not in text, 'TLS must be removed after fallback'
assert ':80 {' in text, 'HTTP must still be served'
print('PASS: reverted to plain HTTP so the site stays reachable\n')
print('=' * 68)
print('CASE C — verify disabled -> trusts the config, no probe')
print('=' * 68)
open(TARGET, 'w').write(':80 {\n respond "x"\n}\n')
hm = load_manager((False, 'should never be called'))
rc = hm._apply(app, https_enabled=True, hostname='digiserver', domain='',
email=None, ip_address='192.168.0.152', port=443,
http_fallback=True, verify=False)
c = config()
print('exit code:', rc, '| https_enabled:', c.https_enabled)
assert rc == 0
assert c.https_enabled is True, 'no probe => config trusted'
assert 'tls internal' in open(TARGET).read()
print('PASS: configuration trusted without probing\n')
print('=' * 68)
print('CASE D — HTTP only (https disabled) is never probed')
print('=' * 68)
open(TARGET, 'w').write(':80 {\n respond "x"\n}\n')
hm = load_manager((False, 'n/a'))
rc = hm._apply(app, https_enabled=False, hostname=None, domain=None,
email=None, ip_address=None, port=443,
http_fallback=True, verify=True)
c = config()
print('exit code:', rc, '| https_enabled:', c.https_enabled)
assert rc == 0
assert c.https_enabled is False
assert 'tls internal' not in open(TARGET).read()
print('PASS: no TLS emitted, no probe performed\n')
shutil.rmtree(TMPDIR, ignore_errors=True)
print('ALL FALLBACK CASES PASSED')
+75
View File
@@ -0,0 +1,75 @@
"""Exercise https_manager.py exactly as deploy.sh does, inside a container-like env.
Overrides the Caddyfile path (the only host-only difference) to prove the
enable/status/disable flow and the resulting Caddyfile content.
"""
import os
import sys
import tempfile
tmp = tempfile.mkdtemp()
os.environ['DATABASE_URL'] = f'sqlite:///{tmp}/t.db'
from app.app import create_app
from app.models.https_config import HTTPSConfig
from app.utils.caddy_manager import CaddyConfigGenerator
app = create_app('production')
with app.app_context():
from app.extensions import db
db.create_all()
# Redirect the Caddyfile write/read to a temp path (the only host difference).
_target = f'{tmp}/Caddyfile'
CaddyConfigGenerator.write_caddyfile = staticmethod(
lambda content, path=_target: (open(_target, 'w').write(content), True)[1])
CaddyConfigGenerator.reload_caddy = staticmethod(lambda: True)
sys.argv = ['https_manager.py', 'enable', 'digiserver', '',
'admin@example.com', '192.168.0.152', '8443']
import importlib.util
spec = importlib.util.spec_from_file_location('hm', '/home/scheianu/digiserver-v2/https_manager.py')
hm = importlib.util.module_from_spec(spec)
spec.loader.exec_module(hm)
print('=== ENABLE (empty domain -> internal CA) ===')
rc = hm.main()
print('exit code:', rc)
print()
print('=== resulting Caddyfile site blocks ===')
text = open(_target).read()
for ln in text.splitlines():
s = ln.strip()
if s.startswith(('http://', 'https://', ':80')) or 'tls internal' in s or 'redir' in s:
print(' ', s)
with app.app_context():
c = HTTPSConfig.get_config()
print()
print('=== stored config ===')
print(' https_enabled:', c.https_enabled)
print(' domain :', repr(c.domain))
print(' ip_address :', c.ip_address)
print(' port :', c.port)
assert c.https_enabled is True
assert not c.domain, 'domain must be empty for internal CA'
assert 'tls internal' in text, 'internal CA directive missing'
assert 'http://192.168.0.152' in text, 'HTTP fallback missing'
print()
print('ASSERT PASS: internal CA + HTTP fallback generated correctly')
print()
print('=== STATUS ===')
sys.argv = ['https_manager.py', 'status']
hm.main()
print()
print('=== DISABLE ===')
sys.argv = ['https_manager.py', 'disable']
hm.main()
with app.app_context():
c = HTTPSConfig.get_config()
print(' https_enabled after disable:', c.https_enabled)
+161
View File
@@ -0,0 +1,161 @@
"""Verify the background player-build flow: state machine + shallow git clone.
Exercises the real functions against a local throwaway bare git repo so no
network is needed, then confirms:
* shallow clone produces a usable working tree
* _run_git never blocks on credentials (GIT_TERMINAL_PROMPT=0)
* a broken/partial checkout is detected and replaced, not reused
* the background job moves idle -> running -> success and records the version
* a bad repo URL ends in 'error', not a hang or a traceback
Run: PYTHONPATH=$(pwd) ./.venv/bin/python docs/tools/test_player_build.py
"""
import os
import shutil
import subprocess
import sys
import tempfile
import time
TMP = tempfile.mkdtemp()
os.environ['DATABASE_URL'] = f'sqlite:///{TMP}/pb.db'
from app.app import create_app # noqa: E402
from app.extensions import db # noqa: E402
from app.utils import player_build as pb # noqa: E402
app = create_app('production')
with app.app_context():
db.create_all()
# ── Build a local origin repo (no network) ──────────────────────────────────
ORIGIN = os.path.join(TMP, 'origin.git')
SRC = os.path.join(TMP, 'src')
subprocess.run(['git', 'init', '--bare', '-q', ORIGIN], check=True)
subprocess.run(['git', 'init', '-q', SRC], check=True)
for k, v in (('user.email', 't@t'), ('user.name', 'T')):
subprocess.run(['git', '-C', SRC, 'config', k, v], check=True)
os.makedirs(os.path.join(SRC, 'config'), exist_ok=True)
open(os.path.join(SRC, 'config', 'app_config.json'), 'w').write('{}')
open(os.path.join(SRC, 'main.py'), 'w').write('print("player")\n')
subprocess.run(['git', '-C', SRC, 'add', '-A'], check=True)
subprocess.run(['git', '-C', SRC, 'commit', '-qm', 'init'], check=True)
subprocess.run(['git', '-C', SRC, 'branch', '-M', 'main'], check=True)
subprocess.run(['git', '-C', SRC, 'remote', 'add', 'origin', ORIGIN], check=True)
subprocess.run(['git', '-C', SRC, 'push', '-q', 'origin', 'main'], check=True)
HEAD = subprocess.run(['git', '-C', SRC, 'rev-parse', '--short', 'HEAD'],
capture_output=True, text=True).stdout.strip()
# file:// forces a real transport so --depth is honoured (a local path clone
# silently ignores it, which would make the shallow assertion meaningless).
ORIGIN_URI = 'file://' + ORIGIN
TARGET = os.path.join(TMP, 'staged')
META = os.path.join(TMP, 'player_build.json')
failures = []
def check(label, cond, detail=''):
print(f" [{'PASS' if cond else 'FAIL'}] {label}" + (f' {detail}' if detail else ''))
if not cond:
failures.append(label)
print('=' * 68)
print('CASE 1 — fresh shallow clone')
print('=' * 68)
r = pb.build_player_files(TARGET, ORIGIN_URI, 'main')
check('clone succeeded', r['success'], r['message'])
check('version matches origin', r['version'] == HEAD, f"{r['version']} vs {HEAD}")
check('working tree has files',
os.path.isfile(os.path.join(TARGET, 'main.py')))
check('checkout reported usable', pb.is_valid_checkout(TARGET))
check('shallow (depth 1)',
os.path.isfile(os.path.join(TARGET, '.git', 'shallow')),
'file:// transport honours --depth')
print()
print('=' * 68)
print('CASE 2 — git never prompts for credentials')
print('=' * 68)
r = pb._run_git(['clone', '--depth', '1',
'https://127.0.0.1:1/nope/nope.git',
os.path.join(TMP, 'nope')], timeout=20)
check('unreachable repo returns fast (no hang)', r.returncode != 0,
f'rc={r.returncode}')
check('GIT_TERMINAL_PROMPT=0 is set', pb._git_env().get('GIT_TERMINAL_PROMPT') == '0')
print()
print('=' * 68)
print('CASE 3 — broken/partial checkout is replaced, not reused')
print('=' * 68)
BROKEN = os.path.join(TMP, 'broken')
shutil.rmtree(BROKEN, ignore_errors=True)
os.makedirs(os.path.join(BROKEN, '.git'), exist_ok=True) # .git but no HEAD
open(os.path.join(BROKEN, 'leftover.txt'), 'w').write('stale')
check('broken dir is not considered usable', not pb.is_valid_checkout(BROKEN))
r = pb.build_player_files(BROKEN, ORIGIN_URI, 'main')
check('build recovers from broken dir', r['success'], r['message'])
check('stale file removed',
not os.path.exists(os.path.join(BROKEN, 'leftover.txt')))
check('now a valid checkout', pb.is_valid_checkout(BROKEN))
print()
print('=' * 68)
print('CASE 4 — background job: idle -> running -> success')
print('=' * 68)
TARGET2 = os.path.join(TMP, 'staged2')
pb._set_build_state(state='idle', step='', message='', version=None)
check('starts idle', pb.get_build_state()['state'] == 'idle')
with app.app_context():
started = pb.start_background_build(
player_code_dir=TARGET2, repo_url=ORIGIN_URI, branch='main',
config_payload={'server_ip': '192.168.0.152', 'port': '443',
'use_https': True, 'verify_ssl': False,
'orientation': 'Landscape', 'max_resolution': '1920x1080'},
meta_path=META, built_by='tester')
check('build accepted', started is True)
check('second start refused while running',
pb.start_background_build(TARGET2, ORIGIN_URI, 'main', None, META, 'x') is False)
for _ in range(120):
if pb.get_build_state()['state'] in ('success', 'error'):
break
time.sleep(0.5)
state = pb.get_build_state()
check('finished successfully', state['state'] == 'success', state.get('message', ''))
check('version recorded', state.get('version') == HEAD)
check('config written to staged code',
os.path.isfile(os.path.join(TARGET2, 'config', 'app_config.json')))
check('build settings persisted', os.path.isfile(META))
settings = pb.load_build_settings(META) or {}
check('meta has server_ip', settings.get('server_ip') == '192.168.0.152',
str(settings.get('server_ip')))
print()
print('=' * 68)
print('CASE 5 — bad repository URL ends in error (no hang, no traceback)')
print('=' * 68)
TARGET3 = os.path.join(TMP, 'staged3')
pb._set_build_state(state='idle', step='', message='', version=None)
with app.app_context():
pb.start_background_build(TARGET3, os.path.join(TMP, 'does-not-exist.git'),
'main', None, META, 'tester')
for _ in range(120):
if pb.get_build_state()['state'] in ('success', 'error'):
break
time.sleep(0.5)
state = pb.get_build_state()
check('reported as error', state['state'] == 'error')
check('error has a message', bool(state.get('message')), state.get('message', '')[:70])
check('no partial dir left behind', not os.path.exists(TARGET3))
shutil.rmtree(TMP, ignore_errors=True)
print()
if failures:
print(f'FAILED: {failures}')
sys.exit(1)
print('ALL PLAYER-BUILD CASES PASSED')
+96
View File
@@ -0,0 +1,96 @@
"""Verify every Caddyfile mode against the real `caddy validate` binary.
Also asserts the structural guarantees required for the deployment model:
* ONE HTTP endpoint that answers regardless of Host header (catch-all :80)
* explicit per-name HTTP blocks for both IP and hostname
* HTTPS on 443 only when enabled, using the internal CA for intranet names
Run: PYTHONPATH=$(pwd) ./.venv/bin/python docs/tools/verify_caddyfile_modes.py
"""
import subprocess
import sys
from app.utils.caddy_manager import CaddyConfigGenerator as G
class Cfg:
def __init__(self, **kw):
self.email = kw.get('email', 'admin@example.com')
self.https_enabled = kw.get('https_enabled', False)
self.domain = kw.get('domain', '')
self.ip_address = kw.get('ip_address', '')
self.hostname = kw.get('hostname', '')
self.port = kw.get('port', 443)
def validate(text):
path = '/tmp/_caddyfile_check'
open(path, 'w').write(text)
r = subprocess.run(
['docker', 'run', '--rm', '-v', f'{path}:/etc/caddy/Caddyfile:ro',
'caddy:2-alpine', 'caddy', 'validate', '--config', '/etc/caddy/Caddyfile'],
capture_output=True, text=True)
return (r.returncode == 0 and 'Valid configuration' in (r.stdout + r.stderr),
(r.stdout + r.stderr))
CASES = [
('1. Nothing configured -> HTTP only on :80',
Cfg(), True, {}),
('2. HTTPS on, IP only -> internal CA + HTTP fallback',
Cfg(https_enabled=True, ip_address='192.168.0.152'), True,
{'tls internal', 'http://192.168.0.152', 'https://192.168.0.152', ':80'}),
('3. HTTPS on, IP + hostname -> both served',
Cfg(https_enabled=True, ip_address='192.168.0.152', hostname='digiserver'),
True,
{'http://digiserver', 'https://digiserver', 'http://192.168.0.152'}),
('4. HTTPS on, redirect-only -> 301 to published port',
Cfg(https_enabled=True, ip_address='192.168.0.152'), False,
{'redir https://192.168.0.152{uri} 301', ':80'}),
('5. HTTPS on, public domain -> ACME (no tls internal)',
Cfg(https_enabled=True, domain='example.com',
ip_address='192.168.0.152'), True,
{'https://example.com', 'tls internal'}),
('6. HTTP only even though IP known (HTTPS disabled) -> :80 only',
Cfg(https_enabled=False, ip_address='192.168.0.152'), True,
{':80'}),
]
failures = []
for label, cfg, fallback, must_contain in CASES:
text = G.generate_caddyfile(cfg, http_fallback=fallback)
ok, raw = validate(text)
missing = sorted(s for s in must_contain if s not in text)
if missing:
ok = False
print(f'[{"OK " if ok else "FAIL"}] {label}')
blocks = [ln.strip() for ln in text.splitlines()
if ln.strip().startswith(('http://', 'https://', ':80'))]
print(f' blocks: {blocks}')
if missing:
print(f' MISSING: {missing}')
if not ok:
failures.append(label)
print(' ', raw.strip()[-400:])
print()
# Extra guarantees that are easy to regress silently.
text_https = G.generate_caddyfile(
Cfg(https_enabled=True, ip_address='192.168.0.152', hostname='digiserver'),
http_fallback=True)
assert ':80 {' in text_https, 'catch-all :80 block missing'
assert 'https://digiserver {' in text_https, 'hostname TLS block missing'
assert text_https.count(':80 {') == 1, 'exactly one catch-all :80 expected'
text_http = G.generate_caddyfile(Cfg(https_enabled=False))
assert 'https://' not in text_http, 'no TLS should be emitted when disabled'
assert ':80 {' in text_http
print('Structural guarantees hold (catch-all :80, per-name blocks, TLS only when enabled)')
if failures:
print(f'FAILED: {failures}')
sys.exit(1)
print('All Caddyfile modes are VALID')
+55
View File
@@ -0,0 +1,55 @@
"""Verify 'legacy code' is excluded from the Docker build context.
Docker's .dockerignore semantics (as applied by BuildKit):
* 'legacy code/' -> directory named "legacy code" at the context root
* '**/legacy code/' -> directory named "legacy code" at ANY depth
This simulates a walk of the build context and asserts the snapshot never
appears in the files that would be sent to the daemon.
"""
import os
ROOT = os.getcwd()
rules = []
for raw in open('.dockerignore', encoding='utf-8'):
line = raw.strip()
if line and not line.startswith('#'):
rules.append(line)
print('rules mentioning legacy:', [r for r in rules if 'legacy' in r])
def is_legacy_dir(rel):
"""True if *rel* is a directory named 'legacy code' at root or any depth."""
return rel == 'legacy code' or rel.endswith('/legacy code')
excluded, included = [], []
for dirpath, dirnames, filenames in os.walk(ROOT):
dirnames[:] = [d for d in dirnames if d != '.git']
for d in list(dirnames):
rel = os.path.relpath(os.path.join(dirpath, d), ROOT)
if is_legacy_dir(rel):
excluded.append(rel + '/ (pruned)')
dirnames.remove(d)
continue
for f in filenames:
rel = os.path.relpath(os.path.join(dirpath, f), ROOT)
parts = rel.split('/')
under_legacy = any(is_legacy_dir('/'.join(parts[:i]))
for i in range(1, len(parts) + 1))
(excluded if under_legacy else included).append(rel)
print()
print('EXCLUDED (legacy snapshot):')
for e in sorted(excluded)[:5]:
print(' -', e)
print(f' ... {len(excluded)} total')
print()
leak = [i for i in included if 'legacy code' in i]
print(f'files reaching the build context: {len(included)}')
print('LEAKED:', leak if leak else 'none')
assert not leak, 'legacy snapshot would ship into the image!'
print()
print('VERIFIED: legacy snapshot is excluded from the Docker build context')