212 lines
9.9 KiB
Python
212 lines
9.9 KiB
Python
"""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.
|
|
'app/runtime.py': (
|
|
'payment = FakePayment()',
|
|
),
|
|
'app/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 / 'app' / 'core' / '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'app/core/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 app/core/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('app/core/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')
|
|
if numeric.get('SCAN_MAX_BYTES', 0) > 128 * 1024 * 1024:
|
|
errors.append('SCAN_MAX_BYTES cannot exceed the configured ClamAV 128 MiB stream limit')
|
|
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))
|