first commit
This commit is contained in:
100
local/staging_readiness.py
Normal file
100
local/staging_readiness.py
Normal file
@@ -0,0 +1,100 @@
|
||||
"""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 local.staging_readiness PATH')
|
||||
raise SystemExit(main(Path(sys.argv[1])))
|
||||
Reference in New Issue
Block a user