"""Dependency-free pre-push checks, also embedded in the Actions executor. This is a regression guard for known shadow stores, not a JavaScript verifier. Diagnostics contain filenames/rules only, never source or credentials. """ from __future__ import annotations import os import hashlib import json import re import sys from pathlib import Path _SKIP = { ".git", ".p2hs", "node_modules", ".nuxt", ".output", ".next", "dist", "build", "coverage", ".venv", "venv", "__pycache__", "tests", "__tests__", } _SKIP_FILES = { "p2hs-codegen-action-runner.mjs", } _SUFFIXES = {".js", ".ts", ".mjs", ".cjs", ".jsx", ".tsx", ".vue"} _SHADOW_MODULES = { "user-db", "records-db", "local-auth", "data-bridge", "app-storage", "auth-seed", } _PLATFORM_MARKERS = ( "p2hs-auth-users", "p2hs-auth-password", "p2hs-records", "p2hs-contact", "p2hs-auth-magic", "p2hs-forms", "p2hs-forms-runtime", "p2hs-forms-submit", ) _RETIRED_MARKER = "__P2HS_RETIRED_LOCAL_STORE__" _RULES = { "local seeded login directory": re.compile(r"usr_seed_"), "local user directory storage": re.compile( r"""(?:useStorage\s*\(\s*["'][^"']*(?:users|auth)|""" r"""(?:localStorage|sessionStorage)\s*\.\s*setItem\s*\(\s*["']""" r"""(?:users|accounts|auth_users)["'])""", re.I, ), "local SQLite database": re.compile( r"""(?:from\s*|require\s*\(\s*|import\s*\(\s*)["'](?:better-sqlite3|sqlite3|node:sqlite)["']""" ), "file-backed database": re.compile( r"""(?:from\s*|require\s*\(\s*|import\s*\(\s*)["']lowdb(?:/[^"']*)?["']""" ), "local business record storage": re.compile( r"""useStorage\s*\(\s*["'](?:db|data|records|orders|products|contacts|bookings)(?:[:/][^"']*)?["']""", re.I, ), "file-backed business records": re.compile( r"""\bwriteFile(?:Sync)?\s*\(\s*["'`][^"'`]*""" r"""(?:users|accounts|records|orders|products|contacts)\.json["'`]""", re.I, ), "missing Cloud config local fallback": re.compile( r"(?:CLOUD_GATEWAY_URL|CLOUD_SERVICE_TOKEN|cloudGatewayUrl|cloudServiceToken)" r".{0,240}" r"(?:useStorage|usr_seed_|SEED_|localUsers|users\.json|new\s+Map\s*\()", re.I | re.DOTALL, ), "seeded local user table": re.compile(r"\bSEED_USERS\b"), "Nitro key used as user directory": re.compile( r"""['"]auth:users['"]""" ), "in-memory persistence success": re.compile( r"serving from in-memory cache|login still works in-memory|" r"durable write skipped|durable local auth fallback", re.I, ), # Unconditional -- bypassing the resolved email capability is wrong regardless of # what runtime-contract.json says (see runtime_contract.py's email precedence). "vendor email client bypassing the platform email capability": re.compile( r"""(?:from\s*|require\s*\(\s*|import\s*\(\s*)["'](?:resend|nodemailer)["']""" r"""|smtp://[^\s"'`]+""", re.I, ), "ORM/database client replacing Cloud records": re.compile( r"""(?:from\s*|require\s*\(\s*|import\s*\(\s*)["'](?:@prisma/client|prisma|mongoose|typeorm|drizzle-orm)["']""", re.I, ), "project-local domain event emission": re.compile( r"""["'`]/api/(?:domain-events|events)(?:/|\?|["'`])""" r"""|\bemitDomainEvent\b|\bemitCanonicalEvent\b""", re.I, ), "project-local workflow enforcement": re.compile( r"""(?:ALLOWED_)?TRANSITIONS\s*=\s*\{[^}]{0,400}(?:reviewing|quoted)""" r"""|\bvalidateWorkflowTransition\s*\(""", re.I, ), } # Only active when runtime-contract.json names a non-default admin path (see # _load_runtime_contract). A literal, complete "/admin" token used as a redirect/next/ # href target is the exact shape of the real magic-link bug this guard now generalizes -- # deliberately narrow so it doesn't flag a longer, still-valid path like # "/admin/reset-password" or an unrelated string that merely contains "/admin". _ADMIN_REDIRECT_CONTEXT = re.compile( r"""(?:location\s*\.\s*(?:replace|assign)\s*\(|location\s*\.\s*href\s*=|""" r"""\bredirect\s*\(|\bnext\s*[:=]|\bhref\s*=)\s*["'`]/admin["'`]""", re.I, ) _HTML_SUFFIXES = {".html", ".htm"} _MAGIC_RULES = { "homemade magic-link landing": re.compile( r"""["'`]/(?:admin|auth)/magic(?:/|\?|["'`])""", re.I, ), } # Whole-file presence, not proximity: real generated staff scripts read ?token= in # one function and POST magic-link/verify in another. That leftover consume must be # migrated onto /api/auth/magic-link/consume (see migrate_legacy_app_paths) rather # than allowed to keep failing later prompts. _MAGIC_LINK_VERIFY_CALL = re.compile(r"magic-link/verify", re.I) _MAGIC_LINK_TOKEN_FROM_URL = re.compile(r"searchParams|location\.search|URLSearchParams", re.I) _MAGIC_CONSUME_MARKER = "__P2HS_MAGIC_CONSUME_MIGRATED__" _HOMEMADE_MAGIC_PATH = re.compile( r"""(["'`])/(?:admin|auth)/magic(?=\1|\?|/)""", re.I, ) # Leftover homemade contact POST / inbox reads from before L1/L2 ports. # Do not match canonical `/api/forms/...` — that rewrite used to reintroduce `/api/contact`. _LEFTOVER_CONTACT_POST = re.compile( r"""(["'`])(?:https?://api\.resend\.com/emails|/api/(?:send-email|send-mail|send(?:-message)?|email|mail|leads|contact-form|submit-contact|submit))(\1)""", re.I, ) _LEFTOVER_CONTACT_INBOX = re.compile( r"""(["'`])(/api/(?:submissions|leads|messages|contacts|contact-submissions|form-submissions|inbox))(\1)""", re.I, ) _COLLECTION = re.compile( r"\b(?:const|let|var)\s+((?:users|accounts|records|orders|products|contacts)(?:Store|Db)?)\s*" r"(?:\:[^=;\n]+)?=\s*(?:new\s+Map\b|\[)", re.I, ) def _stem(name: str) -> str: if name.endswith(".p2hs-local.bak"): name = name[: -len(".p2hs-local.bak")] return Path(name).stem def _load_runtime_contract(root: Path) -> dict | None: """p2hs/runtime-contract.json, written by port_adapter.py at install time. Absent (older project that hasn't regenerated since Phase 4, or contract resolution failed) -> route checks are skipped, not treated as a violation. """ path = root / "p2hs" / "runtime-contract.json" if not path.is_file(): return None try: data = json.loads(path.read_text(encoding="utf-8-sig")) except (OSError, json.JSONDecodeError, UnicodeDecodeError): return None return data if isinstance(data, dict) else None _CLOUD_WRITE_MARKERS = ( "p2hs-contact", "p2hs-forms", "p2hs-records", "/api/forms/", "submitGeneratedForm", "/api/contact", "/api/records", "submitContact", "createRecord", ) _CLOUD_READ_MARKERS = ( "/api/contact/inbox", "listInbox", "/api/records", "listRecords", ) def _skip_platform_or_retired(source: str) -> bool: if "useStorage" in source or "usr_seed_" in source: return False if _RETIRED_MARKER in source: return True return any(marker in source for marker in _PLATFORM_MARKERS) def route_overlaps(rel: str, routes: list[str]) -> bool: """Conservative route overlap across Pages and Nitro static/dynamic/catch-all files.""" if not rel.startswith(("server/api/", "functions/api/")): return False route = rel.split('/api/', 1)[1] route = re.sub(r'\.(?:[cm]?[jt]sx?)$', '', route) route = re.sub(r'\.(?:get|post|put|patch|delete|head|options)$', '', route) parts = [p for p in route.split('/') if not (p.startswith('(') and p.endswith(')'))] if parts[-1] == 'index': parts.pop() for reserved in routes: expected = reserved.removeprefix('/api/').split('/') for i, part in enumerate(parts): if part.startswith('[[') or part.startswith('[...'): return True if i >= len(expected): break if not (part.startswith('[') or expected[i].startswith(':') or part == expected[i]): break else: if len(parts) == len(expected): return True return False def validate_port_artifacts(repo_path: str, expected_lock: dict | None = None) -> None: root = Path(repo_path) lock_path = root / 'p2hs/platform-ports.lock.json' if expected_lock is None and not lock_path.exists(): return # Existing repositories are pinned on their next codegen installation. lock = expected_lock if expected_lock is not None else json.loads(lock_path.read_text(encoding='utf-8')) if expected_lock is not None: if not lock_path.is_file() or json.loads(lock_path.read_text(encoding='utf-8')) != expected_lock: raise ValueError('Platform port lock was changed or removed') for rel, digest in lock['files'].items(): target = root / rel if not target.resolve().is_relative_to(root.resolve()) or target.is_symlink() or not target.is_file(): raise ValueError(f'Platform port artifact missing or unsafe: {rel}') if hashlib.sha256(target.read_bytes()).hexdigest() != digest: raise ValueError(f'Platform port artifact modified: {rel}') for directory in (root / 'server/api', root / 'functions/api'): if not directory.is_dir(): continue for target in directory.rglob('*'): rel = target.relative_to(root).as_posix() if target.is_file() and target.suffix in _SUFFIXES and rel not in lock['files'] and route_overlaps(rel, lock['routes']): raise ValueError(f'Homemade route shadows an installed platform port: {rel}') def _retire_to_p2hs_bak(root: Path, rel: str) -> None: src = root / rel dest = root / ".p2hs/local-bak" / rel dest.parent.mkdir(parents=True, exist_ok=True) if dest.exists(): dest.unlink() src.replace(dest) def _purge_scanned_backups(root: Path) -> None: for folder in ("server", "functions"): base = root / folder if not base.is_dir(): continue for path in base.rglob("*"): if path.is_file() and path.name.endswith(".p2hs-local.bak"): path.unlink() def _consume_redirect_snippet(admin_path: str, *, html: bool) -> str: next_default = admin_path or "/admin" js = ( f"/* {_MAGIC_CONSUME_MARKER} */\n" "(function(){try{" "var p=new URLSearchParams(location.search);var t=p.get('token');" "if(!t)return;" "if((location.pathname||'').indexOf('/api/auth/magic-link/consume')===0)return;" f"var next=(location.pathname||'{next_default}')+(location.hash||'');" "location.replace('/api/auth/magic-link/consume?token='+encodeURIComponent(t)" "+'&next='+encodeURIComponent(next));" "}catch(e){}})();\n" ) if html: return f"\n" return js def migrate_app_source( source: str, *, admin_path: str, html: bool = False, rewrite_legacy_contact_post: bool = False, ) -> str: """Rewrite leftover invented auth landings onto platform consume + the contract admin path.""" if _skip_platform_or_retired(source): return source out = source if rewrite_legacy_contact_post: out = _LEFTOVER_CONTACT_POST.sub(r"\1/api/contact\2", out) out = _LEFTOVER_CONTACT_INBOX.sub(r"\1/api/contact/inbox\3", out) path = admin_path if admin_path.startswith("/") else f"/{admin_path}" if not path or path == "/": path = "/admin" if ( _MAGIC_CONSUME_MARKER not in out and _MAGIC_LINK_VERIFY_CALL.search(out) and _MAGIC_LINK_TOKEN_FROM_URL.search(out) ): out = _consume_redirect_snippet(path, html=html) + out out = _MAGIC_LINK_VERIFY_CALL.sub("magic-link/consume", out) out = _HOMEMADE_MAGIC_PATH.sub(lambda match: f"{match.group(1)}{path}", out) if path != "/admin" and _ADMIN_REDIRECT_CONTEXT.search(out): out = _ADMIN_REDIRECT_CONTEXT.sub( lambda match: re.sub( r"""["'`]/admin["'`]""", lambda quote: f"{quote.group(0)[0]}{path}{quote.group(0)[0]}", match.group(0), ), out, ) return out def migrate_legacy_app_paths(repo_path: str) -> list[str]: """One-shot rewrite of leftover consume / magic / admin / homemade contact URLs.""" root = Path(repo_path) contract = _load_runtime_contract(root) admin_path = ((contract or {}).get("routes") or {}).get("admin_panel_path") or "/admin" # This module is embedded verbatim into the GitHub Actions runner and # executed from a customer repository. Keep the detection dependency-free; # importing Core's ``src`` package works locally but fails in Actions. rewrite_legacy_contact_post = _repo_has_legacy_contact_submit(root) changed: list[str] = [] for directory, dirs, files in os.walk(root, followlinks=False): dirs[:] = [d for d in dirs if d not in _SKIP and not (Path(directory) / d).is_symlink()] for name in files: path = Path(directory) / name inspect_backend = path.suffix in _SUFFIXES inspect_html = path.suffix in _HTML_SUFFIXES if not inspect_backend and not inspect_html: continue if re.search(r"\.(?:test|spec)\.", name) or name in _SKIP_FILES: continue if name.endswith(".p2hs-local.bak") or path.is_symlink(): continue source = path.read_text(encoding="utf-8-sig") updated = migrate_app_source( source, admin_path=admin_path, html=inspect_html, rewrite_legacy_contact_post=rewrite_legacy_contact_post, ) if updated != source: path.write_text(updated, encoding="utf-8", newline="\n") changed.append(path.relative_to(root).as_posix()) return changed def prepare_port_routes(repo_path: str, lock: dict) -> list[str]: """Retire preexisting routes before generation; never run this after the model.""" root = Path(repo_path) backed = [] previous = root / 'p2hs/platform-ports.lock.json' if previous.is_file(): old = json.loads(previous.read_text(encoding='utf-8')) for rel in old.get('files', {}): target = root / rel if (target.resolve().is_relative_to(root.resolve()) and not target.is_symlink() and rel not in lock['files'] and target.is_file()): _retire_to_p2hs_bak(root, rel) backed.append(rel) for directory in (root / 'server/api', root / 'functions/api'): if directory.is_dir(): for target in directory.rglob('*'): rel = target.relative_to(root).as_posix() if target.is_file() and target.suffix in _SUFFIXES and rel not in lock['files'] and route_overlaps(rel, lock['routes']): _retire_to_p2hs_bak(root, rel) backed.append(rel) _purge_scanned_backups(root) backed.extend(migrate_legacy_app_paths(repo_path)) return backed def _repo_has_legacy_contact_submit(root: Path) -> bool: """Detect an existing legacy contact submit route without Core imports.""" for rel in ("functions/api/contact.js", "server/api/contact.post.ts"): if (root / rel).is_file(): return True lock_path = root / "p2hs/platform-ports.lock.json" if not lock_path.is_file(): return False try: lock = json.loads(lock_path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError, TypeError, UnicodeDecodeError): return False if not isinstance(lock, dict): return False if lock.get("legacy_contact_submit") is True: return True files = lock.get("files") or {} if any(rel in files for rel in ("functions/api/contact.js", "server/api/contact.post.ts")): return True return "/api/contact" in (lock.get("routes") or []) def _load_app_block_plan(root: Path) -> dict | None: path = root / "p2hs" / "app-blocks-plan.json" if not path.is_file(): return None try: data = json.loads(path.read_text(encoding="utf-8-sig")) except (OSError, json.JSONDecodeError, UnicodeDecodeError): return None return data if isinstance(data, dict) else None def _locked_port_rels(root: Path) -> set[str]: path = root / "p2hs" / "platform-ports.lock.json" if not path.is_file(): return set() try: data = json.loads(path.read_text(encoding="utf-8-sig")) except (OSError, json.JSONDecodeError, UnicodeDecodeError): return set() files = data.get("files") if isinstance(data, dict) else None if not isinstance(files, dict): return set() return {str(rel) for rel in files} _FORM_BLOCK_IDS = frozenset({"forms.contact", "forms.quote_request"}) # Canonical Forms first. Leftover /api/contact and createRecord remain accepted # compatibility wiring, not required for new authoritative form blocks. _FORM_BIND_MARKERS = ( "submitGeneratedForm", "/api/forms/", "submitContact", "/api/contact", "createRecord", ) def _app_block_binding_violations(root: Path, _scanned_sources: list[str]) -> list[str]: """Authoritative public-form blocks must be wired in app code, not only adapters.""" plan = _load_app_block_plan(root) if not plan or plan.get("adoption") != "authoritative": return [] ids = { item.get("id") for item in (plan.get("blocks") or []) if isinstance(item, dict) } needed = ids & _FORM_BLOCK_IDS if not needed: return [] locked = _locked_port_rels(root) found = False for directory, dirs, files in os.walk(root, followlinks=False): dirs[:] = [d for d in dirs if d not in _SKIP and not (Path(directory) / d).is_symlink()] for name in files: path = Path(directory) / name if path.suffix not in _SUFFIXES and path.suffix not in _HTML_SUFFIXES: continue rel = path.relative_to(root).as_posix() if rel in locked or name in _SKIP_FILES or name.endswith(".p2hs-local.bak"): continue try: source = path.read_text(encoding="utf-8-sig") except (OSError, UnicodeDecodeError): continue if any(marker in source for marker in _FORM_BIND_MARKERS): found = True break if found: break if found: return [] return [ "app_block_binding: authoritative plan requires a public form wired to " "submitGeneratedForm / POST /api/forms/{formKey}/submissions " "(legacy submitContact / POST /api/contact / createRecord still accepted; " "not only platform adapters)" ] def validate_backend_repo(repo_path: str) -> None: """Reject known authoritative local stores before publishing a revision.""" root = Path(repo_path) if not root.is_dir(): raise ValueError("Backend validation requires a repository checkout") validate_port_artifacts(repo_path) # The guard is embedded verbatim into the customer repository's GitHub # Actions runner. That runner does not contain Core's ``src`` package, # so keep the local import for Core execution but use the already-embedded # helpers when running inside the standalone runner. try: from src.domain.codegen.generated_auth_conformance import ( assert_generated_auth_conformance, canonical_generation_plan_required, load_generation_plan_artifact, ) except ModuleNotFoundError as exc: # The Actions runner embeds this module in the generated repository, # where Core's top-level ``src`` package is intentionally absent. # ``exc.name`` is usually ``src`` for that first import, while local # Core execution may report a deeper ``src.domain...`` name. if not exc.name or (exc.name != "src" and not exc.name.startswith("src.")): raise try: assert_generated_auth_conformance = globals()["assert_generated_auth_conformance"] canonical_generation_plan_required = globals()["canonical_generation_plan_required"] load_generation_plan_artifact = globals()["load_generation_plan_artifact"] except KeyError as missing: raise RuntimeError( "Embedded generated-auth conformance helpers are missing" ) from missing generation_plan = load_generation_plan_artifact(root) auth_conformance_violations: list[str] = [] if generation_plan is None and canonical_generation_plan_required(root): auth_conformance_violations.append( "AUTH-GEN-000: canonical generated project is missing " "p2hs/generation-plan.json; artifact is invalid" ) elif generation_plan is not None: assert_generated_auth_conformance(root, generation_plan) contract = _load_runtime_contract(root) admin_path = ((contract or {}).get("routes") or {}).get("admin_panel_path") or "/admin" check_admin_literal = admin_path != "/admin" locked_platform_paths = _locked_port_rels(root) magic_link_allowed = bool( ((contract or {}).get("capabilities") or {}).get("auth", {}).get("magic_link") ) violations = list(auth_conformance_violations) scanned_sources: list[str] = [] for directory, dirs, files in os.walk(root, followlinks=False): dirs[:] = sorted( d for d in dirs if d not in _SKIP and not (Path(directory) / d).is_symlink() ) for name in sorted(files): path = Path(directory) / name inspect_backend = path.suffix in _SUFFIXES inspect_magic = inspect_backend or path.suffix in _HTML_SUFFIXES if not inspect_magic or re.search(r"\.(?:test|spec)\.", name): continue if name in _SKIP_FILES or name.endswith(".p2hs-local.bak"): continue if path.is_symlink(): raise ValueError("Backend validation cannot inspect source symlinks") rel = path.relative_to(root).as_posix() source = path.read_text(encoding="utf-8-sig") scanned_sources.append(source) platform = rel in locked_platform_paths or _skip_platform_or_retired(source) if inspect_backend and not platform: if _stem(name) in _SHADOW_MODULES: violations.append(f"{rel}: local dual-store module") for rule, pattern in _RULES.items(): if pattern.search(source): violations.append(f"{rel}: {rule}") # UI lists are normal. Mutable backend collections are a common fake DB. if any( part in {"server", "functions", "api"} for part in path.relative_to(root).parts[:-1] ): for match in _COLLECTION.finditer(source): if re.search( r"\b" + re.escape(match[1]) + r"\s*\.\s*(?:push|set|splice)\s*\(", source ): violations.append(f"{rel}: mutable local backend collection") if not platform: for rule, pattern in _MAGIC_RULES.items(): if pattern.search(source): violations.append(f"{rel}: {rule}") if contract is not None and not magic_link_allowed and re.search( r"requestMagicLink|/api/auth/magic-link/request|magic_link", source, re.I, ): violations.append( f"{rel}: magic_link_disabled_by_runtime_contract" ) if _MAGIC_LINK_VERIFY_CALL.search(source) and _MAGIC_LINK_TOKEN_FROM_URL.search(source): violations.append(f"{rel}: client-side magic-link consume") if check_admin_literal and _ADMIN_REDIRECT_CONTEXT.search(source): violations.append( f"{rel}: hardcoded /admin redirect (this project's admin path is {admin_path})" ) blob = "\n".join(scanned_sources) reqs = (contract or {}).get("requirements") or {} if isinstance(reqs, dict) and reqs.get("persistence_required"): declared = (root / "p2hs" / "provision.json").is_file() or ( root / "p2hs" / "feature-packs.json" ).is_file() if not declared and not any(marker in blob for marker in _CLOUD_WRITE_MARKERS): violations.append( "persistence_required: no Cloud write path " "(p2hs-contact / p2hs-forms / p2hs-records, /api/contact, /api/records, " "or p2hs/provision.json / p2hs/feature-packs.json)" ) if isinstance(reqs, dict) and reqs.get("admin_must_read_persisted"): if not any(marker in blob for marker in _CLOUD_READ_MARKERS): violations.append( "admin_must_read_persisted: no staff read of persisted Cloud data " "(/api/contact/inbox, listInbox, /api/records, or listRecords)" ) violations.extend(_app_block_binding_violations(root, scanned_sources)) if violations: raise ValueError( "Cloud persistence contract failed. Use platform auth for identities and " "Cloud records for application data; no local fallback. Magic links must " "use /api/auth/magic-link/consume, not a homemade /admin/magic page.\n" + "\n".join(violations[:30]) ) if __name__ == "__main__": try: if "--migrate" in sys.argv: migrate_legacy_app_paths(sys.argv[1] if len(sys.argv) > 1 and sys.argv[1] != "--migrate" else ".") sys.exit(0) if len(sys.argv) > 3 and sys.argv[3] == '--prepare': prepare_port_routes(sys.argv[1], json.loads(sys.stdin.read() if sys.argv[2] == '-' else sys.argv[2])) sys.exit(0) if len(sys.argv) > 2: validate_port_artifacts(sys.argv[1], json.loads(sys.stdin.read() if sys.argv[2] == '-' else sys.argv[2])) validate_backend_repo(sys.argv[1] if len(sys.argv) > 1 else ".") except (ValueError, OSError) as exc: print(str(exc), file=sys.stderr) sys.exit(1)