Files
dtf-system/ops/staging_readiness.py
2026-09-23 10:40:18 -03:00

101 lines
4.2 KiB
Python

"""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])))