Rental-operations-platform/scripts/phase16-static-audit.py

#!/usr/bin/env python3
"""Dependency-free Phase 16 repository integrity, privacy, and release-governance audit."""
from __future__ import annotations

import argparse
import csv
import hashlib
import json
import os
import re
import stat
import sys
from collections import Counter
from pathlib import Path
from typing import Any

EXCLUDED_DIRS = {'.git', '.next', 'node_modules', 'coverage', 'playwright-report', 'test-results', '__pycache__'}
EXCLUDED_AUDIT_OUTPUTS = {
    'docs/quality/PHASE_15_STATIC_AUDIT_RESULTS_v1.0.json',
    'docs/quality/PHASE_15_STATIC_AUDIT_RESULTS_v1.0.md',
    'docs/quality/PHASE_15_BASELINE_STATIC_AUDIT_RESULTS_v1.0.json',
    'docs/quality/PHASE_15_BASELINE_STATIC_AUDIT_RESULTS_v1.0.md',
}
TEXT_SUFFIXES = {
    '.css', '.csv', '.html', '.js', '.json', '.jsx', '.md', '.mjs', '.mts', '.txt', '.ts', '.tsx', '.yml', '.yaml'
}
RESERVED_EMAIL_SUFFIXES = ('.test', '.invalid', '.example')
ALLOWED_EMAIL_DOMAINS = {'example.com', 'example.org', 'example.net'}
SECRET_PATTERNS = {
    'private_key': re.compile(r'-----BEGIN (?:RSA |EC |DSA |OPENSSH )?PRIVATE KEY-----'),
    'aws_access_key': re.compile(r'\bAKIA[0-9A-Z]{16}\b'),
    'google_api_key': re.compile(r'\bAIza[0-9A-Za-z_-]{35}\b'),
    'github_token': re.compile(r'\bgh[pousr]_[0-9A-Za-z]{20,}\b'),
    'slack_token': re.compile(r'\bxox[baprs]-[0-9A-Za-z-]{10,}\b'),
    'stripe_secret': re.compile(r'\bsk_(?:live|test)_[0-9A-Za-z]{16,}\b'),
}
EMAIL_RE = re.compile(r'(?<![\w.+-])([A-Za-z0-9._%+-]+@([A-Za-z0-9.-]+\.[A-Za-z]{2,}))')
RUNTIME_EXTERNAL_URL_RE = re.compile(r'https?://[^\s\)\]"\'<>]+')


def walk_files(root: Path):
    for path in root.rglob('*'):
        if any(part in EXCLUDED_DIRS for part in path.relative_to(root).parts):
            continue
        relative = path.relative_to(root).as_posix()
        if relative in EXCLUDED_AUDIT_OUTPUTS:
            continue
        if path.is_file() or path.is_symlink():
            yield path


def sha256(path: Path) -> str:
    h = hashlib.sha256()
    with path.open('rb') as f:
        for chunk in iter(lambda: f.read(1024 * 1024), b''):
            h.update(chunk)
    return h.hexdigest()


def read_text(path: Path) -> str | None:
    if path.suffix.lower() not in TEXT_SUFFIXES and path.name not in {'.env.example', '.gitignore', '.npmrc', '.nvmrc', '.node-version', '.editorconfig', '.prettierignore', '.dockerignore'}:
        return None
    try:
        return path.read_text(encoding='utf-8')
    except UnicodeDecodeError:
        return None


def add(findings: list[dict[str, Any]], severity: str, code: str, path: str, message: str, line: int | None = None):
    item = {'severity': severity, 'code': code, 'path': path, 'message': message}
    if line is not None:
        item['line'] = line
    findings.append(item)


def audit(root: Path) -> dict[str, Any]:
    findings: list[dict[str, Any]] = []
    metrics: dict[str, Any] = {}
    files = list(walk_files(root))
    regular_files = [p for p in files if p.is_file() and not p.is_symlink()]
    metrics['file_count'] = len(regular_files)
    metrics['total_bytes'] = sum(p.stat().st_size for p in regular_files)
    metrics['suffix_counts'] = dict(sorted(Counter(p.suffix.lower() or '<none>' for p in regular_files).items()))

    # Filesystem integrity.
    for p in files:
        rel = p.relative_to(root).as_posix()
        if p.is_symlink():
            add(findings, 'High', 'symlink_present', rel, f'Symlink target: {os.readlink(p)}')
        elif p.stat().st_mode & stat.S_IWOTH:
            add(findings, 'Medium', 'world_writable', rel, 'Repository file is world-writable.')
    for directory in EXCLUDED_DIRS - {'.git'}:
        if (root / directory).exists():
            add(findings, 'Medium', 'generated_directory_present', directory, 'Generated/cache directory is present in the release package.')
    for forbidden in ('.env', '.env.local', '.env.production', 'package-lock.json', 'tsconfig.tsbuildinfo'):
        if (root / forbidden).exists():
            add(findings, 'High' if forbidden.startswith('.env') else 'Medium', 'forbidden_root_file', forbidden, 'Unexpected generated or secret-bearing root file.')

    # JSON and CSV structural validity.
    json_count = 0
    csv_count = 0
    for p in regular_files:
        rel = p.relative_to(root).as_posix()
        if p.suffix.lower() == '.json':
            json_count += 1
            try:
                json.loads(p.read_text(encoding='utf-8'))
            except Exception as exc:
                add(findings, 'High', 'invalid_json', rel, str(exc))
        elif p.suffix.lower() == '.csv':
            csv_count += 1
            try:
                with p.open(newline='', encoding='utf-8-sig') as f:
                    rows = list(csv.reader(f))
                if rows:
                    width = len(rows[0])
                    for index, row in enumerate(rows, start=1):
                        if len(row) != width:
                            add(findings, 'High', 'malformed_csv', rel, f'Expected {width} columns, found {len(row)}.', index)
            except Exception as exc:
                add(findings, 'High', 'invalid_csv', rel, str(exc))
    metrics['json_files_parsed'] = json_count
    metrics['csv_files_parsed'] = csv_count

    # Secret, PII-fixture, and unsafe runtime scans.
    secret_scan_files = 0
    email_hits: list[dict[str, Any]] = []
    runtime_urls: list[dict[str, Any]] = []
    for p in regular_files:
        rel = p.relative_to(root).as_posix()
        text = read_text(p)
        if text is None:
            continue
        lines = text.splitlines()
        if rel not in {'scripts/phase15-static-audit.py', 'scripts/phase16-static-audit.py'} and not rel.startswith('PHASE_') and p.name != 'pnpm-lock.yaml':
            secret_scan_files += 1
            for name, pattern in SECRET_PATTERNS.items():
                for match in pattern.finditer(text):
                    line = text.count('\n', 0, match.start()) + 1
                    add(findings, 'Critical', f'secret_{name}', rel, 'Potential live credential or private key pattern.', line)
        if not rel.startswith('contracts/phase9/') and p.name != 'pnpm-lock.yaml':
            for index, line_text in enumerate(lines, start=1):
                for match in EMAIL_RE.finditer(line_text):
                    address, domain = match.group(1), match.group(2).lower()
                    email_hits.append({'path': rel, 'line': index, 'address': address})
                    if not domain.endswith(RESERVED_EMAIL_SUFFIXES) and domain not in ALLOWED_EMAIL_DOMAINS:
                        add(findings, 'High', 'non_reserved_email_fixture', rel, f'Non-reserved email address found: {address}', index)
        if rel.startswith('src/') or rel in {'.env.example', 'next.config.ts'}:
            for index, line_text in enumerate(lines, start=1):
                for match in RUNTIME_EXTERNAL_URL_RE.finditer(line_text):
                    url = match.group(0).rstrip('.,;')
                    runtime_urls.append({'path': rel, 'line': index, 'url': url})
                    allowed = (
                        url.startswith('http://localhost')
                        or url.startswith('http://127.0.0.1')
                        or url.startswith('https://your-approved-domain.example')
                        or url.startswith('https://design-tokens.github.io/')
                    )
                    if not allowed:
                        add(findings, 'High', 'unapproved_runtime_url', rel, f'Runtime/config URL is not on the documented safe list: {url}', index)
    metrics['secret_scan_files'] = secret_scan_files
    metrics['email_fixture_hits'] = email_hits
    metrics['runtime_url_hits'] = runtime_urls

    # Runtime anti-patterns and release-control invariants.
    runtime_text = ''
    for p in regular_files:
        rel = p.relative_to(root).as_posix()
        if rel.startswith('src/'):
            text = read_text(p)
            if text is not None:
                runtime_text += f'\n/* {rel} */\n{text}'
                for index, line_text in enumerate(text.splitlines(), start=1):
                    if re.search(r'\b(TODO|TBD|FIXME|lorem ipsum)\b', line_text, re.I):
                        add(findings, 'Medium', 'runtime_placeholder', rel, line_text.strip(), index)
                    if re.search(r'console\.(?:log|info|warn|error|debug)\s*\(', line_text):
                        add(findings, 'High', 'runtime_console_logging', rel, 'Runtime console logging requires privacy review.', index)
                    if re.search(r'\beval\s*\(|new\s+Function\s*\(', line_text):
                        add(findings, 'Critical', 'dynamic_code_execution', rel, 'Dynamic code execution found.', index)
    if "startsWith('application/json')" in runtime_text or 'startsWith("application/json")' in runtime_text:
        add(findings, 'Medium', 'loose_content_type_match', 'src/', 'JSON API still uses a prefix media-type check.')
    helper = root / 'src/lib/security/content-type.ts'
    if not helper.exists() or "=== 'application/json'" not in helper.read_text(encoding='utf-8'):
        add(findings, 'Medium', 'missing_exact_content_type_control', helper.relative_to(root).as_posix(), 'Exact application/json media-type helper is absent.')
    route = root / 'src/app/api/demo/route.ts'
    if route.exists() and 'isApplicationJson(contentType)' not in route.read_text(encoding='utf-8'):
        add(findings, 'Medium', 'content_type_helper_not_used', route.relative_to(root).as_posix(), 'Demo API does not use exact JSON media-type validation.')

    # Visual-test determinism invariants.
    visual_component = root / 'tests/visual/component-system.visual.spec.ts'
    if visual_component.exists():
        text = visual_component.read_text(encoding='utf-8')
        if 'rdg-theme-preference' in text:
            add(findings, 'Medium', 'stale_theme_storage_key', visual_component.relative_to(root).as_posix(), 'Visual test writes a storage key the application never reads.')
        if "hpc.theme.preference" not in text:
            add(findings, 'Medium', 'theme_storage_key_missing', visual_component.relative_to(root).as_posix(), 'Visual test does not use the canonical theme storage key.')
        if "project.name !== 'chromium'" not in text:
            add(findings, 'Medium', 'visual_project_scope', visual_component.relative_to(root).as_posix(), 'Canonical snapshot test is not restricted to Chromium.')
    visual_integrations = root / 'tests/visual/integrations.visual.spec.ts'
    if visual_integrations.exists() and "project.name !== 'chromium'" not in visual_integrations.read_text(encoding='utf-8'):
        add(findings, 'Medium', 'visual_project_scope', visual_integrations.relative_to(root).as_posix(), 'Canonical integration snapshots are not restricted to Chromium.')

    # CI must execute the integration suite; unit and browser suites do not cover the server adapter orchestration.
    workflow_path = root / '.github/workflows/ci.yml'
    if workflow_path.exists():
        workflow_text = workflow_path.read_text(encoding='utf-8')
        if 'pnpm test:integration' not in workflow_text:
            add(findings, 'High', 'ci_integration_suite_omitted', workflow_path.relative_to(root).as_posix(), 'CI does not execute the demo-service integration suite.')

    # Lockfile/direct dependency checks are delegated to the repository validator; record pinning shape here.
    package = json.loads((root / 'package.json').read_text(encoding='utf-8'))
    direct = {**package.get('dependencies', {}), **package.get('devDependencies', {})}
    floating = {name: value for name, value in direct.items() if re.search(r'[~^*xX<>| ]', str(value))}
    if floating:
        add(findings, 'High', 'floating_direct_dependencies', 'package.json', json.dumps(floating, sort_keys=True))
    metrics['direct_dependency_count'] = len(direct)
    metrics['direct_dependencies_exact'] = not floating

    # Issue/gate metrics.
    def issue_version(path: Path) -> tuple[int, ...]:
        match = re.search(r'_v(\d+(?:\.\d+)*)\.csv$', path.name)
        return tuple(int(part) for part in match.group(1).split('.')) if match else (0,)

    issue_candidates = sorted(
        (root / 'docs').glob('IMPLEMENTATION_ISSUE_REGISTER_v*.csv'),
        key=issue_version,
    )
    issue_path = issue_candidates[-1] if issue_candidates else None
    if issue_path is not None:
        with issue_path.open(newline='', encoding='utf-8-sig') as f:
            issues = list(csv.DictReader(f))
        metrics['issue_register_path'] = issue_path.relative_to(root).as_posix()
        metrics['issue_register_rows'] = len(issues)
        unresolved = [r for r in issues if 'resolved' not in (r.get('Status') or '').lower()]
        metrics['unresolved_issue_counts'] = dict(Counter(r.get('Severity', 'Unknown') for r in unresolved))
        if any(None in r for r in issues):
            add(findings, 'High', 'issue_register_overflow', issue_path.relative_to(root).as_posix(), 'One or more issue-register rows contain overflow columns.')
    gate_candidates = [
        root / 'docs/phase16/FINAL_RELEASE_GATE_MATRIX_v1.0.csv',
        root / 'docs/phase15/RELEASE_GATE_MATRIX_v1.0.csv',
        root / 'docs/phase14/RELEASE_GATE_MATRIX_v1.0.csv',
    ]
    gate_path = next((candidate for candidate in gate_candidates if candidate.exists()), None)
    if gate_path is not None:
        with gate_path.open(newline='', encoding='utf-8-sig') as f:
            gates = list(csv.DictReader(f))
        metrics['release_gate_path'] = gate_path.relative_to(root).as_posix()
        metrics['release_gate_status_counts'] = dict(Counter(r.get('Phase16_decision') or r.get('Status', 'Unknown') for r in gates))

    # High-level privacy invariants.
    api_text = route.read_text(encoding='utf-8') if route.exists() else ''
    required_api_controls = {
        'request_marker': "x-rdg-form",
        'origin_validation': 'isAllowedOrigin',
        'body_limit': 'maximumBodyBytes',
        'no_store': "'Cache-Control': 'no-store'",
        'honeypot': 'honeypot',
    }
    for code, token in required_api_controls.items():
        if token not in api_text:
            add(findings, 'High', f'missing_api_{code}', route.relative_to(root).as_posix(), f'Missing expected API control token: {token}')
    analytics_text = (root / 'src/lib/analytics/events.ts').read_text(encoding='utf-8')
    for forbidden in ('email', 'company', 'message', 'phone', 'fieldvalue', 'serverresponse'):
        if f"'{forbidden}'" not in analytics_text:
            add(findings, 'High', 'analytics_forbidden_key_missing', 'src/lib/analytics/events.ts', f'Forbidden analytics key is not explicitly denied: {forbidden}')

    severity_rank = {'Critical': 4, 'High': 3, 'Medium': 2, 'Low': 1, 'Info': 0}
    findings.sort(key=lambda x: (-severity_rank.get(x['severity'], 0), x['path'], x.get('line', 0), x['code']))
    counts = Counter(item['severity'] for item in findings)
    return {
        'schema_version': '1.0',
        'audit': 'Phase 16 dependency-free static repository audit',
        'root': str(root),
        'metrics': metrics,
        'finding_counts': dict(counts),
        'findings': findings,
        'passed': not any(item['severity'] in {'Critical', 'High', 'Medium'} for item in findings),
    }


def render_markdown(result: dict[str, Any]) -> str:
    lines = [
        '# Phase 16 Static Repository Audit',
        '',
        f"- Result: **{'PASS' if result['passed'] else 'FAIL'}**",
        f"- Files inspected: {result['metrics']['file_count']}",
        f"- JSON files parsed: {result['metrics']['json_files_parsed']}",
        f"- CSV files parsed: {result['metrics']['csv_files_parsed']}",
        f"- Secret-scan files: {result['metrics']['secret_scan_files']}",
        f"- Findings: {json.dumps(result['finding_counts'], sort_keys=True)}",
        '',
        '## Findings',
        '',
    ]
    if not result['findings']:
        lines.append('No Critical, High, Medium, Low, or informational findings were produced by this audit.')
    else:
        lines.append('| Severity | Code | Path | Line | Finding |')
        lines.append('|---|---|---|---:|---|')
        for item in result['findings']:
            message = str(item['message']).replace('|', '\\|').replace('\n', ' ')
            lines.append(f"| {item['severity']} | `{item['code']}` | `{item['path']}` | {item.get('line', '')} | {message} |")
    lines.extend(['', '## Metrics', '', '```json', json.dumps(result['metrics'], indent=2, sort_keys=True), '```', ''])
    return '\n'.join(lines)


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument('--root', type=Path, default=Path(__file__).resolve().parents[1])
    parser.add_argument('--json-output', type=Path)
    parser.add_argument('--markdown-output', type=Path)
    args = parser.parse_args()
    result = audit(args.root.resolve())
    if args.json_output:
        args.json_output.parent.mkdir(parents=True, exist_ok=True)
        args.json_output.write_text(json.dumps(result, indent=2, sort_keys=True) + '\n', encoding='utf-8')
    if args.markdown_output:
        args.markdown_output.parent.mkdir(parents=True, exist_ok=True)
        args.markdown_output.write_text(render_markdown(result), encoding='utf-8')
    print(json.dumps({'passed': result['passed'], 'finding_counts': result['finding_counts'], 'files': result['metrics']['file_count']}, sort_keys=True))
    return 0 if result['passed'] else 1


if __name__ == '__main__':
    raise SystemExit(main())