ske087
46602f1933
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.
315 lines
11 KiB
Python
315 lines
11 KiB
Python
"""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()
|