"""Validate non-secret staging decisions without contacting external services.""" from pathlib import Path import re import sys from urllib.parse import urlparse REQUIRED = ( 'APP_ENV', 'STAGING_APPROVED_BY', 'STAGING_PUBLIC_ORIGIN', 'STAGING_S3_ENDPOINT', 'STAGING_S3_BUCKET', 'STAGING_DATABASE_MODE', 'STAGING_SECRET_SOURCE', 'STAGING_BACKUP_DESTINATION', 'STAGING_FREIGHT_PROVIDER', 'STAGING_PAYMENT_PROVIDER', 'STAGING_ERP_PROVIDER', 'STAGING_WHATSAPP_PROVIDER', 'STAGING_ALERT_OWNER', 'STAGING_ROLLBACK_OWNER', ) FORBIDDEN_NAME = re.compile( r'(PASSWORD|TOKEN|SECRET|ACCESS_KEY|PRIVATE_KEY|CREDENTIAL)', re.IGNORECASE) PLACEHOLDERS = {'', 'todo', 'tbd', 'replace-me', 'changeme', 'unconfirmed'} LOCAL_HOSTS = {'localhost', '127.0.0.1', '::1', 'storage', 'db'} def read_config(path: Path) -> dict[str, str]: values = {} for number, raw in enumerate(path.read_text().splitlines(), 1): line = raw.strip() if not line or line.startswith('#'): continue if '=' not in line: raise ValueError(f'{path}:{number}: expected NAME=value') name, value = line.split('=', 1) name = name.strip() if not re.fullmatch(r'[A-Z][A-Z0-9_]*', name): raise ValueError(f'{path}:{number}: invalid setting name') if name in values: raise ValueError(f'{path}:{number}: duplicate setting {name}') if name != 'STAGING_SECRET_SOURCE' and FORBIDDEN_NAME.search(name): raise ValueError(f'{path}:{number}: secrets must not be stored in this file ({name})') values[name] = value.strip() return values def validate(values: dict[str, str]) -> list[str]: errors = [] for name in REQUIRED: if values.get(name, '').lower() in PLACEHOLDERS: errors.append(f'{name} is not decided') if values.get('APP_ENV') != 'staging': errors.append('APP_ENV must be staging') for name in ('STAGING_PUBLIC_ORIGIN', 'STAGING_S3_ENDPOINT'): endpoint = urlparse(values.get(name, '')) if endpoint.scheme != 'https' or not endpoint.hostname: errors.append(f'{name} must be an absolute HTTPS URL') elif endpoint.hostname.lower() in LOCAL_HOSTS: errors.append(f'{name} must not point to localhost or a Compose service') if endpoint.path not in ('', '/') or endpoint.params or endpoint.query or endpoint.fragment: errors.append(f'{name} must not include a path, query, or fragment') storage_host = urlparse(values.get('STAGING_S3_ENDPOINT', '')).hostname or '' if storage_host and not storage_host.endswith('.r2.cloudflarestorage.com'): errors.append('STAGING_S3_ENDPOINT must be a Cloudflare R2 S3 API endpoint') bucket = values.get('STAGING_S3_BUCKET', '') if bucket and not re.fullmatch(r'[a-z0-9][a-z0-9.-]{1,61}[a-z0-9]', bucket): errors.append('STAGING_S3_BUCKET is not a valid S3 bucket name') for name in ('STAGING_FREIGHT_PROVIDER', 'STAGING_PAYMENT_PROVIDER', 'STAGING_ERP_PROVIDER', 'STAGING_WHATSAPP_PROVIDER'): if values.get(name, '').lower() in {'fake', 'local', 'mock'}: errors.append(f'{name} cannot select a local fake provider') if values.get('STAGING_DATABASE_MODE') not in {'managed', 'dedicated-container'}: errors.append('STAGING_DATABASE_MODE must be managed or dedicated-container') if values.get('STAGING_SECRET_SOURCE') in {'env-file', 'repository', '.env'}: errors.append('STAGING_SECRET_SOURCE must be an external secret-injection mechanism') return errors def main(path: Path) -> int: try: errors = validate(read_config(path)) except (OSError, ValueError) as exc: print(f'BLOCKED: {exc}') return 2 if errors: print('BLOCKED: staging inputs are incomplete or unsafe:') for error in errors: print(f'- {error}') return 2 print('PASS: non-secret staging inputs are complete and structurally safe.') print('No provider was contacted. This check does not authorize deployment.') return 0 if __name__ == '__main__': if len(sys.argv) != 2: raise SystemExit('usage: python -m ops.staging_readiness PATH') raise SystemExit(main(Path(sys.argv[1])))