"""Fail closed until the application and non-secret production inputs are ready.""" import os from pathlib import Path import re import sys from urllib.parse import urlparse ROOT = Path(__file__).resolve().parent.parent APPROVALS = ( 'PRODUCTION_DEPLOY_ENABLED', 'PRODUCTION_INPUTS_APPROVED', 'PRODUCTION_SECURITY_REVIEW_APPROVED', 'PRODUCTION_RESTORE_REHEARSED', ) REQUIRED = ( 'API_IMAGE', 'WEB_IMAGE', 'POSTGRES_IMAGE', 'CLAMAV_IMAGE', 'IMAGE_TAG', 'PUBLIC_ORIGIN', 'PUBLIC_HOST', 'KANBAN_HOST', 'SITE_PORT', 'KANBAN_PORT', 'R2_ENDPOINT', 'R2_PUBLIC_ENDPOINT', 'R2_BUCKET', 'POSTGRES_DB', 'POSTGRES_USER', 'APP_DB_USER', 'POSTGRES_VOLUME', 'OPERATOR_EMAIL', 'STORAGE_QUOTA_BYTES', 'OWNER_UPLOAD_QUOTA_BYTES', 'MAX_UPLOAD_BYTES', 'UPLOAD_PART_BYTES', 'MAX_PENDING_UPLOADS', 'SCAN_MAX_BYTES', 'PAYMENT_ADAPTER', 'FREIGHT_ADAPTER', 'TINY_ADAPTER', 'WHATSAPP_ADAPTER', 'DATABASE_URL_SECRET', 'DATABASE_ADMIN_URL_SECRET', 'DB_ADMIN_PASSWORD_SECRET', 'APP_DB_PASSWORD_SECRET', 'R2_ACCESS_KEY_ID_SECRET', 'R2_SECRET_ACCESS_KEY_SECRET', 'OPERATOR_PASSWORD_SECRET', 'PAYMENT_TOKEN_SECRET', 'PAYMENT_WEBHOOK_SECRET', 'TINY_TOKEN_SECRET', 'WHATSAPP_TOKEN_SECRET', ) IMMUTABLE_IMAGE = re.compile(r'^[a-z0-9][a-z0-9._:/-]*@sha256:([0-9a-f]{64})$') IMAGE_REPOSITORY = re.compile( r'^[a-z0-9][a-z0-9.-]*(?::[0-9]{1,5})?(?:/[a-z0-9][a-z0-9._-]*)+$') DNS = re.compile(r'^(?=.{1,253}$)(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)+[a-z]{2,63}$') NAME = re.compile(r'^[a-zA-Z0-9][a-zA-Z0-9_.-]{2,127}$') SOURCE_BLOCKERS = { # Markers must name something that is still true, or the gate weakens without # failing. Four entries here described a local-only runtime and stopped # matching when R2 support landed; they were removed rather than left to rot. # What remains is the real blocker: no production payment or messaging adapter # exists, so these lines must change before a release can be meaningful. 'local/runtime.py': ( 'payment = FakePayment()', ), 'local/worker.py': ( "adapters = {'tiny': FakeTiny(), 'whatsapp': FakeWhatsApp()}", ), } def source_errors(root=ROOT): errors = [] for relative, markers in SOURCE_BLOCKERS.items(): text = (root / relative).read_text() for marker in markers: if marker in text: errors.append(f'{relative} remains local-only: {marker}') errors.extend(secret_loading_errors(root)) return errors def secret_loading_errors(root=ROOT): """Exercise the secret loader instead of grepping it. Searching for a string passes as soon as someone writes that string, and fails when a working implementation happens to spell it differently. Load the module and make it resolve a real file. """ import importlib.util import tempfile module_path = root / 'local' / 'secrets.py' if not module_path.exists(): return ['local runtime does not load the production Docker secret *_FILE settings'] try: spec = importlib.util.spec_from_file_location('_preflight_secrets', module_path) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) except Exception as exc: return [f'local/secrets.py could not be loaded: {exc}'] stack_names = set() stack = root / 'deploy' / 'stack.yaml' if stack.exists(): for line in stack.read_text().splitlines(): if '_FILE:' in line: key = line.split(':')[0].strip() # The database image consumes this one; the application does not. if key and key != 'POSTGRES_PASSWORD_FILE': stack_names.add(key[:-len('_FILE')]) failures = [] with tempfile.TemporaryDirectory() as directory: for name in sorted(stack_names): path = Path(directory) / name path.write_text('resolved-value\n', encoding='utf-8') environ = {f'{name}_FILE': str(path)} try: module.load(environ) except Exception as exc: failures.append(f'{name}_FILE is not resolved by local/secrets.py: {exc}') continue if environ.get(name) != 'resolved-value': failures.append(f'{name}_FILE did not produce {name}') # A missing secret must stop the service, never start it unconfigured. try: module.load({'DATABASE_URL_FILE': str(Path(directory) / 'absent')}) except Exception: pass else: failures.append('local/secrets.py does not fail closed on an unreadable secret') return failures def config_errors(values): errors = [] for name in APPROVALS: if values.get(name) != 'approved': errors.append(f'{name} must equal approved') for name in REQUIRED: value = values.get(name, '') if not value or value.lower() in {'tbd', 'todo', 'replace-me', 'unconfirmed'}: errors.append(f'{name} is missing or unresolved') for name in ('API_IMAGE', 'WEB_IMAGE'): if not IMAGE_REPOSITORY.fullmatch(values.get(name, '')): errors.append(f'{name} must be a lowercase registry repository without a tag') for name in ('POSTGRES_IMAGE', 'CLAMAV_IMAGE'): match = IMMUTABLE_IMAGE.fullmatch(values.get(name, '')) if not match or match.group(1) == '0' * 64: errors.append(f'{name} must be an immutable non-placeholder digest reference') image_tag = values.get('IMAGE_TAG', '') if image_tag != 'latest' and not re.fullmatch(r'(?:[0-9a-f]{40}|[0-9a-f]{64})', image_tag): errors.append('IMAGE_TAG must be latest or a full commit SHA published by Gitea') origin = urlparse(values.get('PUBLIC_ORIGIN', '')) if (origin.scheme != 'https' or not origin.hostname or origin.path not in ('', '/') or origin.params or origin.query or origin.fragment): errors.append('PUBLIC_ORIGIN must be an HTTPS origin without a path') for name in ('PUBLIC_HOST', 'KANBAN_HOST'): if not DNS.fullmatch(values.get(name, '')): errors.append(f'{name} must be a valid lowercase DNS hostname') if origin.hostname and values.get('PUBLIC_HOST') != origin.hostname: errors.append('PUBLIC_ORIGIN hostname must equal PUBLIC_HOST') for name in ('R2_ENDPOINT', 'R2_PUBLIC_ENDPOINT'): endpoint = urlparse(values.get(name, '')) host = endpoint.hostname or '' if (endpoint.scheme != 'https' or not host.endswith('.r2.cloudflarestorage.com') or endpoint.path not in ('', '/') or endpoint.query or endpoint.fragment): errors.append(f'{name} must be an HTTPS Cloudflare R2 S3 API endpoint') bucket = values.get('R2_BUCKET', '') if bucket and not re.fullmatch(r'[a-z0-9][a-z0-9.-]{1,61}[a-z0-9]', bucket): errors.append('R2_BUCKET must be a valid S3 bucket name') for name in ('PAYMENT_ADAPTER', 'FREIGHT_ADAPTER', 'TINY_ADAPTER', 'WHATSAPP_ADAPTER'): if values.get(name, '').lower() in {'', 'fake', 'mock', 'local'}: errors.append(f'{name} must select an approved non-fake implementation') for name in REQUIRED: if name.endswith('_SECRET') and values.get(name) and not NAME.fullmatch(values[name]): errors.append(f'{name} must name a pre-provisioned external Swarm secret') secret_names = [values.get(name, '') for name in REQUIRED if name.endswith('_SECRET')] populated_secret_names = [name for name in secret_names if name] if len(populated_secret_names) != len(set(populated_secret_names)): errors.append('Every production secret setting must use a distinct external Swarm secret') if values.get('POSTGRES_USER') == values.get('APP_DB_USER'): errors.append('Database administrator and runtime user must differ') if values.get('PUBLIC_HOST') == values.get('KANBAN_HOST'): errors.append('Site and Kanban hosts must differ') numeric = {} for name in ( 'STORAGE_QUOTA_BYTES', 'OWNER_UPLOAD_QUOTA_BYTES', 'MAX_UPLOAD_BYTES', 'UPLOAD_PART_BYTES', 'MAX_PENDING_UPLOADS', 'SCAN_MAX_BYTES'): try: numeric[name] = int(values.get(name, '0')) if numeric[name] <= 0: raise ValueError except ValueError: errors.append(f'{name} must be a positive integer') if numeric.get('OWNER_UPLOAD_QUOTA_BYTES', 0) > numeric.get('STORAGE_QUOTA_BYTES', 0): errors.append('OWNER_UPLOAD_QUOTA_BYTES cannot exceed STORAGE_QUOTA_BYTES') if numeric.get('SCAN_MAX_BYTES', 0) > numeric.get('MAX_UPLOAD_BYTES', 0): errors.append('SCAN_MAX_BYTES cannot exceed MAX_UPLOAD_BYTES') ports = {} for name in ('SITE_PORT', 'KANBAN_PORT'): try: ports[name] = int(values.get(name, '0')) if not 1024 <= ports[name] <= 65535: raise ValueError except ValueError: errors.append(f'{name} must be an unprivileged TCP port from 1024 to 65535') if ports.get('SITE_PORT') == ports.get('KANBAN_PORT'): errors.append('SITE_PORT and KANBAN_PORT must differ') return errors def main(source_only=False): errors = source_errors() if not source_only: errors.extend(config_errors(os.environ)) if errors: print('BLOCKED: production preflight failed:') for error in errors: print(f'- {error}') return 2 print('PASS: production source and non-secret deployment metadata passed preflight.') print('This does not replace staging acceptance, image scanning, or human approval.') return 0 if __name__ == '__main__': if len(sys.argv) > 2 or (len(sys.argv) == 2 and sys.argv[1] != '--source-only'): raise SystemExit('usage: python deploy/production_preflight.py [--source-only]') raise SystemExit(main(len(sys.argv) == 2))