1452 lines
60 KiB
Python
1452 lines
60 KiB
Python
#!/usr/bin/env python3
|
|
"""Validate V2 production configuration without starting or changing services."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import ipaddress
|
|
import json
|
|
import math
|
|
import os
|
|
import re
|
|
import ssl
|
|
import stat
|
|
from collections import Counter
|
|
from dataclasses import asdict, dataclass, field
|
|
from datetime import datetime, timezone
|
|
from email.headerregistry import Address
|
|
from pathlib import Path
|
|
from urllib.parse import urlparse
|
|
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[2]
|
|
DEFAULT_ARTIFACT_ROOT = REPO_ROOT / "project-docs" / "qa-artifacts" / "v2-production-preflight"
|
|
IMMUTABLE_IMAGE_PATTERN = re.compile(r"^[^\s]+@sha256:[0-9a-fA-F]{64}$")
|
|
IDENTIFIER_PATTERN = re.compile(r"^[a-zA-Z_][a-zA-Z0-9_]*$")
|
|
RELEASE_ID_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")
|
|
RUNTIME_TARGET_ID_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")
|
|
SHA256_PATTERN = re.compile(r"^[0-9a-f]{64}$")
|
|
IMAGE_DIGEST_PATTERN = re.compile(r"^sha256:[0-9a-f]{64}$")
|
|
PLACEHOLDER_MARKERS = ("change_me", "local-only", "example.com", "your-api-key", "replace_with")
|
|
LOOPBACK_VALUES = {"127.0.0.1", "::1", "localhost"}
|
|
PROVIDER_ACCEPTANCE_RECEIPT_MAX_BYTES = 1024 * 1024
|
|
PROVIDER_ACCEPTANCE_KEYRING_MAX_BYTES = 256 * 1024
|
|
PUBLIC_POLICY_CANDIDATE_MAX_BYTES = 2 * 1024 * 1024
|
|
PUBLIC_POLICY_APPROVAL_MAX_BYTES = 256 * 1024
|
|
PUBLIC_POLICY_KEYRING_MAX_BYTES = 256 * 1024
|
|
EXPECTED_PUBLIC_POLICY_IDS = {"user_terms", "privacy_policy", "service_agreement", "ai_use_notice"}
|
|
ALERT_WEBHOOK_MAX_BYTES = 16 * 1024
|
|
TLS_CERTIFICATE_MAX_BYTES = 1024 * 1024
|
|
TLS_PRIVATE_KEY_MAX_BYTES = 256 * 1024
|
|
TLS_TRUST_BUNDLE_MAX_BYTES = 2 * 1024 * 1024
|
|
SMTP_CREDENTIAL_CONTAINER_PATH = "/run/secrets/contract_smtp_password"
|
|
LAUNCH_SCOPE_MANIFEST_PATH = (
|
|
REPO_ROOT / "project-docs" / "V2-production-launch-scope-2026-08-12.json"
|
|
)
|
|
FORBIDDEN_SUPPORT_ACCESS_ACTIVATION_KEYS = frozenset(
|
|
{
|
|
"CONTRACT_API_IDENTITY_EXECUTOR_ENABLED",
|
|
"CONTRACT_API_IDENTITY_WRITE_API_ENABLED",
|
|
"CONTRACT_API_PRIVILEGED_IDENTITY_EXECUTOR_ENABLED",
|
|
"CONTRACT_API_PRIVILEGED_IDENTITY_WRITE_API_ENABLED",
|
|
"CONTRACT_API_SUPPORT_ACCESS_ENABLED",
|
|
"CONTRACT_API_SUPPORT_ACCESS_EXECUTOR_ENABLED",
|
|
"CONTRACT_API_SUPPORT_ACCESS_GRANT_ENABLED",
|
|
"CONTRACT_API_SUPPORT_ACCESS_IMPERSONATION_ENABLED",
|
|
"CONTRACT_API_SUPPORT_ACCESS_REVOKE_ENABLED",
|
|
}
|
|
)
|
|
EXPECTED_LAUNCH_SCOPE = {
|
|
"schema_version": 1,
|
|
"decision_id": "support-access-first-release-exclusion-2026-08-12",
|
|
"release_scope": "v2-first-production-release",
|
|
"decision": "excluded",
|
|
"support_access": {
|
|
"included": False,
|
|
"enabled": False,
|
|
"kill_switch_engaged": True,
|
|
"executor_enabled": False,
|
|
},
|
|
"identity_executor": {"enabled": False},
|
|
"destructive_identity_write_api": {
|
|
"included": False,
|
|
"forbidden_operations": [
|
|
"execute",
|
|
"grant",
|
|
"revoke",
|
|
"impersonate",
|
|
"impersonation",
|
|
],
|
|
},
|
|
"future_enablement_gates": [
|
|
"external_idp",
|
|
"mfa_step_up",
|
|
"session_device_binding",
|
|
"real_resource_toctou",
|
|
"cross_node_revocation_sla",
|
|
"kill_switch_drill",
|
|
"immutable_per_request_audit",
|
|
"privacy_legal_approval",
|
|
],
|
|
"notes": [
|
|
"This is a first-release scope exclusion, not approval of support access.",
|
|
"No shared password, manual bypass, identity write API, or execute API may restore this capability.",
|
|
],
|
|
}
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Check:
|
|
key: str
|
|
status: str
|
|
message: str
|
|
details: dict[str, object] = field(default_factory=dict)
|
|
|
|
|
|
def parse_env_file(path: Path) -> dict[str, str]:
|
|
values: dict[str, str] = {}
|
|
for line_number, raw_line in enumerate(path.read_text(encoding="utf-8-sig").splitlines(), start=1):
|
|
line = raw_line.strip()
|
|
if not line or line.startswith("#"):
|
|
continue
|
|
if line.startswith("export "):
|
|
line = line[7:].lstrip()
|
|
if "=" not in line:
|
|
raise ValueError(f"Invalid env syntax at line {line_number}.")
|
|
key, value = line.split("=", 1)
|
|
key = key.strip()
|
|
if not IDENTIFIER_PATTERN.fullmatch(key):
|
|
raise ValueError(f"Invalid env key at line {line_number}.")
|
|
value = value.strip()
|
|
if len(value) >= 2 and value[0] == value[-1] and value[0] in {"'", '"'}:
|
|
value = value[1:-1]
|
|
values[key] = value
|
|
return values
|
|
|
|
|
|
def validate_environment(values: dict[str, str], *, template_mode: bool = False) -> list[Check]:
|
|
checks: list[Check] = []
|
|
|
|
required = (
|
|
"COMPOSE_PROJECT_NAME",
|
|
"RELEASE_ID",
|
|
"YUQEI_RELEASE_ENVIRONMENT_ID",
|
|
"YUQEI_RELEASE_ENVIRONMENT_CLASS",
|
|
"YUQEI_RELEASE_HOST_ID",
|
|
"CONTRACT_API_IMAGE",
|
|
"CONTRACT_WEB_IMAGE",
|
|
"AI_PLATFORM_API_IMAGE",
|
|
"AI_PLATFORM_RELEASE_IMAGE_DIGEST",
|
|
"AI_WORKER_IMAGE",
|
|
"BACKEND_SERVER_NAME",
|
|
"BACKEND_PUBLIC_BASE_URL",
|
|
"BACKEND_TLS_CERT_DIR",
|
|
"BACKEND_TLS_TRUSTED_CA_FILE",
|
|
"BACKUP_HOST_DIR",
|
|
"AI_PROVIDER_API_KEY_FILE",
|
|
"CONTRACT_API_REDIS_URL",
|
|
"AI_PLATFORM_REDIS_URL",
|
|
"YUQEI_WORKER_REDIS_URL",
|
|
"POSTGRES_USER",
|
|
"POSTGRES_PASSWORD",
|
|
"CONTRACT_POSTGRES_DB",
|
|
"AI_PLATFORM_POSTGRES_DB",
|
|
"CONTRACT_API_DATABASE_URL",
|
|
"CONTRACT_API_AI_PLATFORM_TIMEOUT_SECONDS",
|
|
"AI_PLATFORM_DATABASE_URL",
|
|
"AI_PLATFORM_PROVIDER_REQUEST_TIMEOUT_SECONDS",
|
|
"YUQEI_WORKER_DATABASE_URL",
|
|
"MINIO_ROOT_USER",
|
|
"MINIO_ROOT_PASSWORD",
|
|
"CONTRACT_API_WEB_PROXY_SHARED_SECRET",
|
|
"CONTRACT_API_IDENTITY_RATE_LIMIT_SECRET",
|
|
"CONTRACT_API_ANTI_ABUSE_HASH_SECRET",
|
|
"CONTRACT_WEB_SESSION_SECRET",
|
|
"AI_PLATFORM_SERVICE_SHARED_SECRET",
|
|
"AI_PLATFORM_ANTI_ABUSE_HASH_SECRET",
|
|
"AI_PLATFORM_PROVIDER_ACCEPTANCE_RECEIPT_FILE_HOST",
|
|
"AI_PLATFORM_PROVIDER_ACCEPTANCE_TRUSTED_KEYS_FILE_HOST",
|
|
"AI_PLATFORM_PROVIDER_ACCEPTANCE_EXPECTED_RECEIPT_SHA256",
|
|
"PUBLIC_POLICY_CANDIDATE_DIR_HOST",
|
|
"PUBLIC_POLICY_LEGAL_APPROVAL_FILE_HOST",
|
|
"PUBLIC_POLICY_TRUSTED_KEYRING_FILE_HOST",
|
|
"CONTRACT_API_PUBLIC_POLICY_RELEASE_DIGEST",
|
|
"PRODUCTION_ALERT_WEBHOOK_URL_FILE",
|
|
)
|
|
missing = [key for key in required if not values.get(key, "").strip()]
|
|
checks.append(_result("required_values", not missing, "All required values are present." if not missing else f"Missing required keys: {', '.join(missing)}."))
|
|
|
|
exact_values = {
|
|
"CONTRACT_API_ENVIRONMENT": "production",
|
|
"CONTRACT_API_DATABASE_AUTO_CREATE": "false",
|
|
"CONTRACT_API_SEED_DEMO_DATA": "false",
|
|
"CONTRACT_REVIEW_WORKER_SEED_DEMO_DATA": "false",
|
|
"CONTRACT_API_REQUIRE_WEB_PROXY_AUTH": "true",
|
|
"CONTRACT_API_ANTI_ABUSE_MODE": "enforce",
|
|
"CONTRACT_API_PUBLIC_POLICY_EXECUTOR_ENABLED": "false",
|
|
"CONTRACT_WEB_SESSION_SECURE": "true",
|
|
"AI_PLATFORM_ENVIRONMENT": "production",
|
|
"AI_PLATFORM_DATABASE_AUTO_CREATE": "false",
|
|
"AI_PLATFORM_REPOSITORY_BACKEND": "sqlalchemy",
|
|
"AI_PLATFORM_PROVIDER_CALL_MODE": "enabled",
|
|
"AI_PLATFORM_PROVIDER_FALLBACK_ENABLED": "false",
|
|
"AI_PLATFORM_REQUIRE_SERVICE_AUTH": "true",
|
|
"AI_PLATFORM_ANTI_ABUSE_MODE": "enforce",
|
|
"YUQEI_WORKER_ENVIRONMENT": "production",
|
|
"YUQEI_WORKER_DATABASE_AUTO_CREATE": "false",
|
|
}
|
|
invalid_exact = [
|
|
f"{key}={values.get(key, '<missing>')}"
|
|
for key, expected in exact_values.items()
|
|
if values.get(key, "").strip() != expected
|
|
]
|
|
checks.append(_result("production_modes", not invalid_exact, "Production modes are strict." if not invalid_exact else f"Invalid production values: {', '.join(invalid_exact)}."))
|
|
target_ids = {
|
|
"YUQEI_RELEASE_ENVIRONMENT_ID": values.get(
|
|
"YUQEI_RELEASE_ENVIRONMENT_ID", ""
|
|
).strip(),
|
|
"YUQEI_RELEASE_HOST_ID": values.get("YUQEI_RELEASE_HOST_ID", "").strip(),
|
|
}
|
|
invalid_target_ids = [
|
|
key
|
|
for key, value in target_ids.items()
|
|
if not RUNTIME_TARGET_ID_PATTERN.fullmatch(value)
|
|
]
|
|
environment_class = values.get("YUQEI_RELEASE_ENVIRONMENT_CLASS", "").strip()
|
|
target_ok = not invalid_target_ids and environment_class in {
|
|
"isolated_uat",
|
|
"production",
|
|
}
|
|
checks.append(
|
|
_result(
|
|
"release_runtime_target",
|
|
target_ok,
|
|
(
|
|
"Release environment and host identity are explicit."
|
|
if target_ok
|
|
else "Release environment_id/host_id must be stable identifiers and "
|
|
"environment_class must be isolated_uat or production."
|
|
),
|
|
)
|
|
)
|
|
checks.append(_validate_identity_email_delivery(values))
|
|
checks.append(_validate_ai_timeout_budget(values))
|
|
checks.append(_validate_capacity_thresholds(values))
|
|
checks.extend(_validate_launch_scope_exclusion(values))
|
|
checks.extend(_validate_anti_abuse_configuration(values, template_mode=template_mode))
|
|
|
|
placeholders = [
|
|
key
|
|
for key, value in values.items()
|
|
if value and any(marker in value.lower() for marker in PLACEHOLDER_MARKERS)
|
|
and not _allowed_example_value(key, values)
|
|
]
|
|
checks.append(_result("placeholders", template_mode or not placeholders, "No unresolved placeholders remain." if not placeholders else f"Unresolved placeholder keys: {', '.join(sorted(placeholders))}.", warning=template_mode and bool(placeholders)))
|
|
|
|
image_keys = (
|
|
"CONTRACT_API_IMAGE",
|
|
"CONTRACT_WEB_IMAGE",
|
|
"AI_PLATFORM_API_IMAGE",
|
|
"AI_WORKER_IMAGE",
|
|
"POSTGRES_IMAGE",
|
|
"POSTGRES_CLIENT_IMAGE",
|
|
"REDIS_IMAGE",
|
|
"MINIO_IMAGE",
|
|
"MINIO_MC_IMAGE",
|
|
"NGINX_IMAGE",
|
|
"BACKUP_UTILITY_IMAGE",
|
|
)
|
|
mutable_images = [key for key in image_keys if not IMMUTABLE_IMAGE_PATTERN.fullmatch(values.get(key, ""))]
|
|
checks.append(_result("immutable_images", template_mode or not mutable_images, "Application images are pinned by digest." if not mutable_images else f"Images must use @sha256 digests: {', '.join(mutable_images)}.", warning=template_mode and bool(mutable_images)))
|
|
|
|
configured_ai_digest = values.get("AI_PLATFORM_RELEASE_IMAGE_DIGEST", "").strip()
|
|
ai_image = values.get("AI_PLATFORM_API_IMAGE", "").strip()
|
|
deployed_ai_digest = ai_image.rsplit("@", 1)[1] if "@" in ai_image else ""
|
|
image_binding_ok = (
|
|
bool(IMAGE_DIGEST_PATTERN.fullmatch(configured_ai_digest))
|
|
and configured_ai_digest == deployed_ai_digest.lower()
|
|
)
|
|
checks.append(
|
|
_result(
|
|
"provider_promotion_image_binding",
|
|
template_mode or image_binding_ok,
|
|
"Provider promotion image digest matches the immutable AI Platform image."
|
|
if image_binding_ok
|
|
else "AI_PLATFORM_RELEASE_IMAGE_DIGEST must exactly match the digest in AI_PLATFORM_API_IMAGE.",
|
|
warning=template_mode and not image_binding_ok,
|
|
)
|
|
)
|
|
|
|
secret_keys = (
|
|
"POSTGRES_PASSWORD",
|
|
"MINIO_ROOT_PASSWORD",
|
|
"CONTRACT_API_WEB_PROXY_SHARED_SECRET",
|
|
"CONTRACT_API_IDENTITY_RATE_LIMIT_SECRET",
|
|
"CONTRACT_API_ANTI_ABUSE_HASH_SECRET",
|
|
"CONTRACT_WEB_SESSION_SECRET",
|
|
"AI_PLATFORM_SERVICE_SHARED_SECRET",
|
|
"AI_PLATFORM_ANTI_ABUSE_HASH_SECRET",
|
|
)
|
|
weak_secrets = [key for key in secret_keys if len(values.get(key, "")) < 32]
|
|
secret_values = [values[key] for key in secret_keys if values.get(key)]
|
|
duplicate_secret_count = len(secret_values) - len(set(secret_values))
|
|
secret_ok = not weak_secrets and duplicate_secret_count == 0
|
|
runtime_message = "Protected runtime values are long and independent."
|
|
if not secret_ok:
|
|
runtime_message = f"Weak protected keys: {', '.join(weak_secrets) or '-'}; duplicate protected values: {duplicate_secret_count}."
|
|
checks.append(_result("runtime_secrets", template_mode or secret_ok, runtime_message, warning=template_mode and not secret_ok))
|
|
|
|
bind_keys = (
|
|
"POSTGRES_BIND_ADDRESS",
|
|
"REDIS_BIND_ADDRESS",
|
|
"AI_REDIS_BIND_ADDRESS",
|
|
"MINIO_API_BIND_ADDRESS",
|
|
"MINIO_CONSOLE_BIND_ADDRESS",
|
|
"CONTRACT_API_BIND_ADDRESS",
|
|
"AI_PLATFORM_API_BIND_ADDRESS",
|
|
)
|
|
public_internal_binds = [key for key in bind_keys if values.get(key, "") not in LOOPBACK_VALUES]
|
|
checks.append(_result("internal_port_binding", not public_internal_binds, "Database, cache, storage, and application ports bind only to loopback." if not public_internal_binds else f"Internal ports are not loopback-bound: {', '.join(public_internal_binds)}."))
|
|
|
|
database_checks = _validate_database_urls(values)
|
|
checks.extend(database_checks)
|
|
checks.extend(_validate_provider_acceptance_deployment(values, template_mode=template_mode))
|
|
checks.extend(_validate_public_policy_deployment(values, template_mode=template_mode))
|
|
|
|
public_url = values.get("BACKEND_PUBLIC_BASE_URL", "")
|
|
parsed_public_url = urlparse(public_url)
|
|
endpoint_ok = parsed_public_url.scheme == "https" and parsed_public_url.hostname == values.get("BACKEND_SERVER_NAME") and not parsed_public_url.path.rstrip("/")
|
|
checks.append(_result("https_endpoint", endpoint_ok, "Backend public URL is an origin-only HTTPS endpoint matching BACKEND_SERVER_NAME." if endpoint_ok else "BACKEND_PUBLIC_BASE_URL must be an HTTPS origin matching BACKEND_SERVER_NAME."))
|
|
|
|
release_id = values.get("RELEASE_ID", "")
|
|
release_ok = bool(RELEASE_ID_PATTERN.fullmatch(release_id)) and not _contains_placeholder(release_id)
|
|
checks.append(_result("release_id", template_mode or release_ok, "Release ID is immutable and explicit." if release_ok else "RELEASE_ID must be a non-placeholder tag or commit identifier.", warning=template_mode and not release_ok))
|
|
|
|
volume_keys = ("POSTGRES_VOLUME_NAME", "MINIO_VOLUME_NAME", "CONTRACT_FILES_VOLUME_NAME")
|
|
invalid_volumes = [key for key in volume_keys if "smoke" in values.get(key, "").lower() or not values.get(key)]
|
|
checks.append(_result("production_volumes", not invalid_volumes, "Stable production volume names are configured." if not invalid_volumes else f"Invalid production volume names: {', '.join(invalid_volumes)}."))
|
|
|
|
backup_value = values.get("BACKUP_HOST_DIR", "")
|
|
backup_root = Path(backup_value or ".")
|
|
is_absolute_host_path = backup_root.is_absolute() or backup_value.startswith("/")
|
|
under_application_root = backup_value == "/opt/apps" or backup_value.startswith("/opt/apps/")
|
|
backup_location_ok = is_absolute_host_path and "smoke" not in backup_value.lower() and not under_application_root
|
|
checks.append(_result("backup_location", backup_location_ok, "Backup destination is absolute and outside the application directory." if backup_location_ok else "BACKUP_HOST_DIR must be absolute, non-smoke, and outside /opt/apps."))
|
|
|
|
return checks
|
|
|
|
|
|
def _validate_anti_abuse_configuration(
|
|
values: dict[str, str],
|
|
*,
|
|
template_mode: bool,
|
|
) -> list[Check]:
|
|
expected_urls = {
|
|
"CONTRACT_API_REDIS_URL": "redis://redis:6379/0",
|
|
"AI_PLATFORM_REDIS_URL": "redis://ai-redis:6379/0",
|
|
}
|
|
url_keys = tuple(expected_urls)
|
|
targets: dict[str, tuple[str, int, int]] = {}
|
|
invalid_urls: list[str] = []
|
|
for key in url_keys:
|
|
target = _parse_redis_target(values.get(key, ""))
|
|
if target is None or values.get(key, "") != expected_urls[key]:
|
|
invalid_urls.append(key)
|
|
else:
|
|
targets[key] = target
|
|
redis_endpoints = {(host, port) for host, port, _database in targets.values()}
|
|
worker_url = values.get("YUQEI_WORKER_REDIS_URL", "")
|
|
worker_ok = worker_url == "redis://ai-redis:6379/1"
|
|
if not worker_ok:
|
|
invalid_urls.append("YUQEI_WORKER_REDIS_URL")
|
|
redis_ok = not invalid_urls and len(redis_endpoints) == len(url_keys)
|
|
redis_message = (
|
|
"Contract and AI anti-abuse use the two exact, independently deployed Compose Redis services, while the worker uses ai-redis database 1; live reachability remains a deployment probe."
|
|
if redis_ok
|
|
else (
|
|
"The production profile requires exact Redis URLs redis://redis:6379/0, redis://ai-redis:6379/0, and worker redis://ai-redis:6379/1; aliases, userinfo, query/fragment components, alternate databases, and external targets are blocked until instance identity can be verified; "
|
|
f"invalid keys: {', '.join(invalid_urls) or 'duplicate Redis target'}."
|
|
)
|
|
)
|
|
|
|
secret_keys = (
|
|
"CONTRACT_API_ANTI_ABUSE_HASH_SECRET",
|
|
"AI_PLATFORM_ANTI_ABUSE_HASH_SECRET",
|
|
)
|
|
weak_secrets = [
|
|
key for key in secret_keys if not _is_high_entropy_secret(values.get(key, ""))
|
|
]
|
|
configured_secrets = [values.get(key, "") for key in secret_keys]
|
|
independent = len(set(configured_secrets)) == len(secret_keys)
|
|
secrets_ok = not weak_secrets and independent
|
|
secrets_message = (
|
|
"Contract and AI anti-abuse use independent high-entropy hashing secrets."
|
|
if secrets_ok
|
|
else (
|
|
f"Weak anti-abuse secrets: {', '.join(weak_secrets) or '-'}; "
|
|
f"independent={str(independent).lower()}."
|
|
)
|
|
)
|
|
return [
|
|
_result(
|
|
"anti_abuse_redis_targets",
|
|
template_mode or redis_ok,
|
|
redis_message,
|
|
warning=template_mode and not redis_ok,
|
|
),
|
|
_result(
|
|
"anti_abuse_secrets",
|
|
template_mode or secrets_ok,
|
|
secrets_message,
|
|
warning=template_mode and not secrets_ok,
|
|
),
|
|
]
|
|
|
|
|
|
def _parse_redis_target(value: str) -> tuple[str, int, int] | None:
|
|
raw = value.strip()
|
|
if not raw or raw != value or _contains_control_character(raw):
|
|
return None
|
|
parsed = urlparse(raw)
|
|
if parsed.scheme not in {"redis", "rediss"} or not parsed.hostname:
|
|
return None
|
|
hostname = parsed.hostname.lower().rstrip(".")
|
|
if not hostname or parsed.fragment:
|
|
return None
|
|
if hostname == "localhost" or hostname.endswith(".localhost"):
|
|
return None
|
|
try:
|
|
address = ipaddress.ip_address(hostname)
|
|
except ValueError:
|
|
if re.fullmatch(r"(?:0[xX][0-9A-Fa-f]+|[0-9.]+)", hostname):
|
|
return None
|
|
try:
|
|
hostname = hostname.encode("idna").decode("ascii").lower()
|
|
except UnicodeError:
|
|
return None
|
|
dns_name = re.fullmatch(
|
|
r"(?=.{1,253}\Z)(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)*"
|
|
r"[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?",
|
|
hostname,
|
|
)
|
|
if dns_name is None:
|
|
return None
|
|
else:
|
|
mapped = getattr(address, "ipv4_mapped", None)
|
|
if (
|
|
address.is_loopback
|
|
or address.is_unspecified
|
|
or address.is_link_local
|
|
or (mapped is not None and mapped.is_loopback)
|
|
):
|
|
return None
|
|
hostname = address.compressed
|
|
try:
|
|
port = parsed.port
|
|
except ValueError:
|
|
return None
|
|
if port is None or not 1 <= port <= 65535:
|
|
return None
|
|
if not re.fullmatch(r"/[0-9]+", parsed.path):
|
|
return None
|
|
database = int(parsed.path[1:])
|
|
if not 0 <= database <= 15:
|
|
return None
|
|
return hostname, port, database
|
|
|
|
|
|
def _is_high_entropy_secret(value: str) -> bool:
|
|
if (
|
|
len(value) < 32
|
|
or len(value) > 4096
|
|
or value != value.strip()
|
|
or _contains_control_character(value)
|
|
or _contains_placeholder(value)
|
|
):
|
|
return False
|
|
counts = Counter(value)
|
|
if len(counts) < 12:
|
|
return False
|
|
length = len(value)
|
|
shannon_entropy = -sum(
|
|
(count / length) * math.log2(count / length) for count in counts.values()
|
|
)
|
|
return shannon_entropy >= 3.0
|
|
|
|
|
|
def validate_files(values: dict[str, str], *, template_mode: bool = False) -> list[Check]:
|
|
if template_mode:
|
|
return [
|
|
Check(
|
|
"protected_files",
|
|
"warning",
|
|
"Template mode skipped TLS, provider key, Provider acceptance receipt/keyring, public-policy evidence, optional SMTP secret, and backup mount checks.",
|
|
)
|
|
]
|
|
checks: list[Check] = []
|
|
checks.extend(_validate_tls_files(values))
|
|
|
|
provider_key = Path(values.get("AI_PROVIDER_API_KEY_FILE", ""))
|
|
provider_key_ok = provider_key.is_file() and 0 < provider_key.stat().st_size <= 64 * 1024
|
|
checks.append(_result("provider_key_file", provider_key_ok, "Provider API key file is present." if provider_key_ok else "AI provider key file is missing, empty, or unexpectedly large."))
|
|
if provider_key_ok and os.name != "nt":
|
|
mode = stat.S_IMODE(provider_key.stat().st_mode)
|
|
secure_mode = mode & 0o077 == 0
|
|
checks.append(_result("provider_key_permissions", secure_mode, "Provider key file is owner-only." if secure_mode else f"Provider key file mode {oct(mode)} exposes group or other permissions."))
|
|
|
|
receipt_path = Path(values.get("AI_PLATFORM_PROVIDER_ACCEPTANCE_RECEIPT_FILE_HOST", ""))
|
|
receipt_ok = _is_bounded_absolute_file(receipt_path, PROVIDER_ACCEPTANCE_RECEIPT_MAX_BYTES)
|
|
checks.append(
|
|
_result(
|
|
"provider_acceptance_receipt_file",
|
|
receipt_ok,
|
|
"The pinned Provider acceptance receipt file is present."
|
|
if receipt_ok
|
|
else "The Provider acceptance receipt must be an absolute, non-empty file no larger than 1 MiB.",
|
|
)
|
|
)
|
|
expected_receipt_hash = values.get("AI_PLATFORM_PROVIDER_ACCEPTANCE_EXPECTED_RECEIPT_SHA256", "")
|
|
receipt_hash_ok = (
|
|
receipt_ok
|
|
and bool(SHA256_PATTERN.fullmatch(expected_receipt_hash))
|
|
and hashlib.sha256(receipt_path.read_bytes()).hexdigest() == expected_receipt_hash
|
|
)
|
|
checks.append(
|
|
_result(
|
|
"provider_acceptance_receipt_hash",
|
|
receipt_hash_ok,
|
|
"The Provider acceptance receipt matches its deployment SHA-256 pin."
|
|
if receipt_hash_ok
|
|
else "The Provider acceptance receipt content does not match its deployment SHA-256 pin.",
|
|
)
|
|
)
|
|
|
|
trusted_keys_path = Path(values.get("AI_PLATFORM_PROVIDER_ACCEPTANCE_TRUSTED_KEYS_FILE_HOST", ""))
|
|
trusted_keys_ok = _is_bounded_absolute_file(
|
|
trusted_keys_path,
|
|
PROVIDER_ACCEPTANCE_KEYRING_MAX_BYTES,
|
|
)
|
|
checks.append(
|
|
_result(
|
|
"provider_acceptance_trusted_keys_file",
|
|
trusted_keys_ok,
|
|
"The Provider acceptance public-key ring is present."
|
|
if trusted_keys_ok
|
|
else "The Provider acceptance public-key ring must be an absolute, non-empty file no larger than 256 KiB.",
|
|
)
|
|
)
|
|
|
|
checks.extend(_validate_public_policy_files(values))
|
|
|
|
email_mode = values.get("CONTRACT_API_IDENTITY_EMAIL_DELIVERY_MODE", "").strip().lower()
|
|
smtp_username = values.get("CONTRACT_API_IDENTITY_SMTP_USERNAME", "").strip()
|
|
if email_mode == "smtp" and smtp_username:
|
|
smtp_password_file = Path(values.get("CONTRACT_API_IDENTITY_SMTP_PASSWORD_FILE_HOST", ""))
|
|
smtp_password_ok = smtp_password_file.is_file() and 0 < smtp_password_file.stat().st_size <= 16 * 1024
|
|
checks.append(
|
|
_result(
|
|
"smtp_password_file",
|
|
smtp_password_ok,
|
|
"SMTP password file is present."
|
|
if smtp_password_ok
|
|
else "SMTP password file is missing, empty, or unexpectedly large.",
|
|
)
|
|
)
|
|
if smtp_password_ok and os.name != "nt":
|
|
mode = stat.S_IMODE(smtp_password_file.stat().st_mode)
|
|
secure_mode = mode & 0o077 == 0
|
|
checks.append(
|
|
_result(
|
|
"smtp_password_permissions",
|
|
secure_mode,
|
|
"SMTP password file is owner-only."
|
|
if secure_mode
|
|
else f"SMTP password file mode {oct(mode)} exposes group or other permissions.",
|
|
)
|
|
)
|
|
else:
|
|
checks.append(
|
|
_result(
|
|
"smtp_password_file",
|
|
True,
|
|
"SMTP password file is not required for the selected identity email mode.",
|
|
)
|
|
)
|
|
|
|
backup_root = Path(values.get("BACKUP_HOST_DIR", ""))
|
|
require_mount = values.get("BACKUP_REQUIRE_SEPARATE_MOUNT", "true").lower() == "true"
|
|
backup_mount_ok = backup_root.is_dir() and (not require_mount or os.path.ismount(backup_root))
|
|
checks.append(_result("backup_mount", backup_mount_ok, "Backup destination is an available separate mount." if backup_mount_ok else "Backup destination is missing or is not a separate mount."))
|
|
|
|
webhook_path = Path(values.get("PRODUCTION_ALERT_WEBHOOK_URL_FILE", ""))
|
|
webhook_ok = _is_bounded_absolute_file(webhook_path, ALERT_WEBHOOK_MAX_BYTES)
|
|
webhook_url = ""
|
|
if webhook_ok:
|
|
try:
|
|
webhook_url = webhook_path.read_text(encoding="utf-8").strip()
|
|
except (OSError, UnicodeError):
|
|
webhook_ok = False
|
|
parsed_webhook = urlparse(webhook_url)
|
|
webhook_ok = webhook_ok and parsed_webhook.scheme == "https" and bool(parsed_webhook.hostname)
|
|
checks.append(
|
|
_result(
|
|
"alert_webhook_file",
|
|
webhook_ok,
|
|
"Production alert webhook file is an owner-only HTTPS URL."
|
|
if webhook_ok
|
|
else "Production alert webhook file must be an absolute, owner-only file containing an HTTPS URL.",
|
|
)
|
|
)
|
|
if webhook_ok and os.name != "nt":
|
|
mode = stat.S_IMODE(webhook_path.stat().st_mode)
|
|
secure_mode = mode & 0o077 == 0
|
|
checks.append(
|
|
_result(
|
|
"alert_webhook_permissions",
|
|
secure_mode,
|
|
"Production alert webhook file is owner-only."
|
|
if secure_mode
|
|
else f"Alert webhook file mode {oct(mode)} exposes group or other permissions.",
|
|
)
|
|
)
|
|
return checks
|
|
|
|
|
|
def _validate_tls_files(values: dict[str, str]) -> list[Check]:
|
|
tls_dir = Path(values.get("BACKEND_TLS_CERT_DIR", ""))
|
|
certificate_path = tls_dir / "tls.crt"
|
|
private_key_path = tls_dir / "tls.key"
|
|
files_ok = _is_secure_tls_file(
|
|
certificate_path,
|
|
TLS_CERTIFICATE_MAX_BYTES,
|
|
) and _is_secure_tls_file(
|
|
private_key_path,
|
|
TLS_PRIVATE_KEY_MAX_BYTES,
|
|
private=True,
|
|
)
|
|
checks = [
|
|
_result(
|
|
"tls_files",
|
|
files_ok,
|
|
"TLS certificate chain and private key are bounded, regular, single-link files without symlink or reparse indirection."
|
|
if files_ok
|
|
else "TLS files must be absolute, bounded, regular, single-link files; symlink/reparse inputs and insecure key ownership or modes are rejected.",
|
|
)
|
|
]
|
|
if not files_ok:
|
|
return checks + _failed_tls_semantic_checks("TLS files could not be inspected.")
|
|
|
|
configured_ca = values.get("BACKEND_TLS_TRUSTED_CA_FILE", "").strip()
|
|
trusted_ca_path = Path(configured_ca) if configured_ca else None
|
|
trusted_ca_ok = trusted_ca_path is not None and _is_secure_tls_file(
|
|
trusted_ca_path, TLS_TRUST_BUNDLE_MAX_BYTES
|
|
)
|
|
checks.append(
|
|
_result(
|
|
"tls_trust_bundle",
|
|
trusted_ca_ok,
|
|
"TLS trust uses an explicit bounded, regular, single-link CA bundle."
|
|
if trusted_ca_ok
|
|
else "BACKEND_TLS_TRUSTED_CA_FILE is required and must be an absolute, bounded, regular, single-link file without symlink/reparse indirection.",
|
|
)
|
|
)
|
|
if not trusted_ca_ok:
|
|
return checks + _failed_tls_semantic_checks("TLS trust material could not be inspected.")
|
|
assert trusted_ca_path is not None
|
|
|
|
try:
|
|
certificate_bytes = certificate_path.read_bytes()
|
|
private_key_bytes = private_key_path.read_bytes()
|
|
trusted_ca_bytes = trusted_ca_path.read_bytes() if trusted_ca_path else None
|
|
except OSError:
|
|
return checks + _failed_tls_semantic_checks("TLS material changed or became unreadable during validation.")
|
|
|
|
binding_details: dict[str, object] = {
|
|
"certificate_sha256": hashlib.sha256(certificate_bytes).hexdigest(),
|
|
"private_key_sha256": hashlib.sha256(private_key_bytes).hexdigest(),
|
|
"trusted_ca_sha256": (
|
|
hashlib.sha256(trusted_ca_bytes).hexdigest()
|
|
if trusted_ca_bytes is not None
|
|
else "os-trust-store"
|
|
),
|
|
"server_name": values.get("BACKEND_SERVER_NAME", "").strip(),
|
|
}
|
|
checks.append(
|
|
_result(
|
|
"tls_material_binding",
|
|
True,
|
|
"TLS certificate, private key, trust source, and server name are SHA-256 bound into the preflight artifact.",
|
|
details=binding_details,
|
|
)
|
|
)
|
|
|
|
key_matches = False
|
|
server_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
|
|
server_context.minimum_version = ssl.TLSVersion.TLSv1_2
|
|
try:
|
|
server_context.load_cert_chain(
|
|
certfile=str(certificate_path),
|
|
keyfile=str(private_key_path),
|
|
password=_reject_encrypted_tls_key,
|
|
)
|
|
key_matches = True
|
|
except (OSError, ssl.SSLError, ValueError):
|
|
key_matches = False
|
|
checks.append(
|
|
_result(
|
|
"tls_certificate_key_match",
|
|
key_matches,
|
|
"TLS certificate public key matches tls.key."
|
|
if key_matches
|
|
else "TLS certificate public key does not match tls.key.",
|
|
)
|
|
)
|
|
server_name = values.get("BACKEND_SERVER_NAME", "").strip()
|
|
handshake_ok = False
|
|
handshake_error: str | None = None
|
|
if key_matches and server_name and not _contains_control_character(server_name):
|
|
try:
|
|
_verify_tls_handshake(
|
|
server_context=server_context,
|
|
server_name=server_name,
|
|
trusted_ca_path=trusted_ca_path,
|
|
)
|
|
handshake_ok = True
|
|
except (OSError, ssl.SSLError, ValueError) as exc:
|
|
handshake_error = type(exc).__name__
|
|
checks.append(
|
|
_result(
|
|
"tls_certificate_hostname",
|
|
handshake_ok,
|
|
"A verified in-memory TLS handshake proves the certificate SAN covers BACKEND_SERVER_NAME."
|
|
if handshake_ok
|
|
else "TLS hostname verification failed during an in-memory client/server handshake.",
|
|
details={"error_type": handshake_error} if handshake_error else {},
|
|
)
|
|
)
|
|
checks.append(
|
|
_result(
|
|
"tls_certificate_validity",
|
|
handshake_ok,
|
|
"A verified in-memory TLS handshake proves the presented chain is currently valid."
|
|
if handshake_ok
|
|
else "TLS certificate validity could not be proven by the in-memory verified handshake.",
|
|
details={"error_type": handshake_error} if handshake_error else {},
|
|
)
|
|
)
|
|
checks.append(
|
|
_result(
|
|
"tls_certificate_chain",
|
|
handshake_ok,
|
|
"A verified in-memory TLS handshake proves the full chain against the explicitly pinned trust bundle."
|
|
if handshake_ok
|
|
else "TLS chain verification failed against the explicitly pinned trust bundle.",
|
|
details={"error_type": handshake_error} if handshake_error else {},
|
|
)
|
|
)
|
|
try:
|
|
stable = (
|
|
hashlib.sha256(certificate_path.read_bytes()).hexdigest()
|
|
== binding_details["certificate_sha256"]
|
|
and hashlib.sha256(private_key_path.read_bytes()).hexdigest()
|
|
== binding_details["private_key_sha256"]
|
|
and hashlib.sha256(trusted_ca_path.read_bytes()).hexdigest()
|
|
== binding_details["trusted_ca_sha256"]
|
|
and _is_secure_tls_file(certificate_path, TLS_CERTIFICATE_MAX_BYTES)
|
|
and _is_secure_tls_file(
|
|
private_key_path, TLS_PRIVATE_KEY_MAX_BYTES, private=True
|
|
)
|
|
and _is_secure_tls_file(trusted_ca_path, TLS_TRUST_BUNDLE_MAX_BYTES)
|
|
)
|
|
except OSError:
|
|
stable = False
|
|
checks.append(
|
|
_result(
|
|
"tls_material_stability",
|
|
stable,
|
|
"TLS material remained unchanged and securely addressed throughout validation."
|
|
if stable
|
|
else "TLS material changed or became indirectly addressed during validation; preflight is blocked.",
|
|
)
|
|
)
|
|
return checks
|
|
|
|
|
|
def _failed_tls_semantic_checks(message: str) -> list[Check]:
|
|
return [
|
|
_result(key, False, message)
|
|
for key in (
|
|
"tls_certificate_key_match",
|
|
"tls_certificate_hostname",
|
|
"tls_certificate_validity",
|
|
"tls_certificate_chain",
|
|
)
|
|
]
|
|
|
|
|
|
def _is_secure_tls_file(path: Path, max_bytes: int, *, private: bool = False) -> bool:
|
|
try:
|
|
if not path.is_absolute():
|
|
return False
|
|
reparse_flag = getattr(stat, "FILE_ATTRIBUTE_REPARSE_POINT", 0)
|
|
for parent in path.parents:
|
|
parent_metadata = os.lstat(parent)
|
|
if stat.S_ISLNK(parent_metadata.st_mode) or (
|
|
reparse_flag
|
|
and getattr(parent_metadata, "st_file_attributes", 0) & reparse_flag
|
|
):
|
|
return False
|
|
metadata = os.lstat(path)
|
|
if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISREG(metadata.st_mode):
|
|
return False
|
|
if reparse_flag and getattr(metadata, "st_file_attributes", 0) & reparse_flag:
|
|
return False
|
|
if metadata.st_nlink != 1 or not 0 < metadata.st_size <= max_bytes:
|
|
return False
|
|
if os.name != "nt":
|
|
allowed_owners = {0, os.geteuid()}
|
|
if metadata.st_uid not in allowed_owners:
|
|
return False
|
|
mode = stat.S_IMODE(metadata.st_mode)
|
|
if mode & 0o022 or (private and mode & 0o077):
|
|
return False
|
|
return True
|
|
except OSError:
|
|
return False
|
|
|
|
|
|
def _verify_tls_handshake(
|
|
*,
|
|
server_context: ssl.SSLContext,
|
|
server_name: str,
|
|
trusted_ca_path: Path | None,
|
|
) -> None:
|
|
client_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
|
|
client_context.minimum_version = ssl.TLSVersion.TLSv1_2
|
|
client_context.check_hostname = True
|
|
client_context.verify_mode = ssl.CERT_REQUIRED
|
|
if trusted_ca_path is None:
|
|
raise ValueError("an explicit TLS trust bundle is required")
|
|
client_context.load_verify_locations(cafile=str(trusted_ca_path))
|
|
|
|
client_in = ssl.MemoryBIO()
|
|
client_out = ssl.MemoryBIO()
|
|
server_in = ssl.MemoryBIO()
|
|
server_out = ssl.MemoryBIO()
|
|
client = client_context.wrap_bio(
|
|
client_in,
|
|
client_out,
|
|
server_side=False,
|
|
server_hostname=server_name,
|
|
)
|
|
server = server_context.wrap_bio(server_in, server_out, server_side=True)
|
|
client_done = False
|
|
server_done = False
|
|
for _attempt in range(32):
|
|
if not client_done:
|
|
try:
|
|
client.do_handshake()
|
|
client_done = True
|
|
except ssl.SSLWantReadError:
|
|
pass
|
|
client_bytes = client_out.read()
|
|
if client_bytes:
|
|
server_in.write(client_bytes)
|
|
|
|
if not server_done:
|
|
try:
|
|
server.do_handshake()
|
|
server_done = True
|
|
except ssl.SSLWantReadError:
|
|
pass
|
|
server_bytes = server_out.read()
|
|
if server_bytes:
|
|
client_in.write(server_bytes)
|
|
if client_done and server_done:
|
|
return
|
|
raise ssl.SSLError("in-memory TLS handshake did not complete")
|
|
|
|
|
|
def _reject_encrypted_tls_key() -> str:
|
|
raise ValueError("encrypted TLS private keys are not supported by the unattended edge runtime")
|
|
|
|
|
|
def _validate_public_policy_files(values: dict[str, str]) -> list[Check]:
|
|
checks: list[Check] = []
|
|
candidate_root = Path(values.get("PUBLIC_POLICY_CANDIDATE_DIR_HOST", ""))
|
|
candidate_ok = candidate_root.is_absolute() and candidate_root.is_dir() and not candidate_root.is_symlink()
|
|
total_size = 0
|
|
symlink_found = False
|
|
if candidate_ok:
|
|
try:
|
|
for path in candidate_root.rglob("*"):
|
|
if path.is_symlink():
|
|
symlink_found = True
|
|
elif path.is_file():
|
|
total_size += path.stat().st_size
|
|
except OSError:
|
|
candidate_ok = False
|
|
candidate_ok = candidate_ok and not symlink_found and 0 < total_size <= PUBLIC_POLICY_CANDIDATE_MAX_BYTES
|
|
checks.append(
|
|
_result(
|
|
"public_policy_candidate_dir",
|
|
candidate_ok,
|
|
"The public-policy candidate directory is absolute, bounded, and contains no symlinks."
|
|
if candidate_ok
|
|
else "The public-policy candidate directory must be absolute, non-empty, no larger than 2 MiB, and free of symlinks.",
|
|
)
|
|
)
|
|
|
|
manifest_ok = False
|
|
if candidate_ok:
|
|
manifest_path = candidate_root / "manifest.json"
|
|
try:
|
|
manifest = _read_strict_json_object(manifest_path)
|
|
documents = manifest.get("documents")
|
|
manifest_ok = (
|
|
manifest.get("schema_version") == "yuqei-public-policy-candidate-v1"
|
|
and manifest.get("existing_user_reconsent") == "required_on_next_login"
|
|
and isinstance(documents, list)
|
|
and len(documents) == 4
|
|
and {item.get("policy_id") for item in documents if isinstance(item, dict)}
|
|
== EXPECTED_PUBLIC_POLICY_IDS
|
|
)
|
|
if manifest_ok:
|
|
for item in documents:
|
|
source_file = item.get("source_file")
|
|
relative = Path(str(source_file))
|
|
resolved = (candidate_root / relative).resolve()
|
|
if (
|
|
not isinstance(source_file, str)
|
|
or not source_file.strip()
|
|
or relative.is_absolute()
|
|
or ".." in relative.parts
|
|
or not resolved.is_relative_to(candidate_root.resolve())
|
|
or not resolved.is_file()
|
|
or resolved.stat().st_size <= 0
|
|
):
|
|
manifest_ok = False
|
|
break
|
|
except (OSError, UnicodeError, ValueError, TypeError):
|
|
manifest_ok = False
|
|
checks.append(
|
|
_result(
|
|
"public_policy_manifest",
|
|
manifest_ok,
|
|
"The candidate manifest binds all four approved public-policy source files."
|
|
if manifest_ok
|
|
else "The candidate manifest must bind exactly user terms, privacy policy, service agreement, and AI notice with safe in-directory sources.",
|
|
)
|
|
)
|
|
|
|
approval_path = Path(values.get("PUBLIC_POLICY_LEGAL_APPROVAL_FILE_HOST", ""))
|
|
approval_ok = _is_bounded_absolute_file(approval_path, PUBLIC_POLICY_APPROVAL_MAX_BYTES)
|
|
if approval_ok:
|
|
try:
|
|
_read_strict_json_object(approval_path)
|
|
except (OSError, UnicodeError, ValueError):
|
|
approval_ok = False
|
|
checks.append(
|
|
_result(
|
|
"public_policy_legal_approval_file",
|
|
approval_ok,
|
|
"The named public-policy legal approval file is present and JSON-valid."
|
|
if approval_ok
|
|
else "The public-policy legal approval must be an absolute JSON file no larger than 256 KiB.",
|
|
)
|
|
)
|
|
|
|
keyring_path = Path(values.get("PUBLIC_POLICY_TRUSTED_KEYRING_FILE_HOST", ""))
|
|
keyring_ok = _is_bounded_absolute_file(keyring_path, PUBLIC_POLICY_KEYRING_MAX_BYTES)
|
|
if keyring_ok:
|
|
try:
|
|
keyring = _read_strict_json_object(keyring_path)
|
|
trust_basis = keyring.get("trust_basis")
|
|
keyring_ok = (
|
|
isinstance(trust_basis, dict)
|
|
and trust_basis.get("policy_activation_authorized") is False
|
|
)
|
|
except (OSError, UnicodeError, ValueError):
|
|
keyring_ok = False
|
|
checks.append(
|
|
_result(
|
|
"public_policy_trusted_keyring_file",
|
|
keyring_ok,
|
|
"The trusted public-policy keyring is JSON-valid and does not itself authorize activation."
|
|
if keyring_ok
|
|
else "The trusted public-policy keyring must be JSON-valid and keep policy_activation_authorized=false.",
|
|
)
|
|
)
|
|
|
|
if os.name != "nt":
|
|
for key, path, enabled in (
|
|
("public_policy_legal_approval_permissions", approval_path, approval_ok),
|
|
("public_policy_trusted_keyring_permissions", keyring_path, keyring_ok),
|
|
):
|
|
if enabled:
|
|
mode = stat.S_IMODE(path.stat().st_mode)
|
|
secure_mode = mode & 0o077 == 0
|
|
checks.append(
|
|
_result(
|
|
key,
|
|
secure_mode,
|
|
"The public-policy evidence file is owner-only."
|
|
if secure_mode
|
|
else f"Public-policy evidence file mode {oct(mode)} exposes group or other permissions.",
|
|
)
|
|
)
|
|
return checks
|
|
|
|
|
|
def _read_strict_json_object(path: Path) -> dict[str, object]:
|
|
value = json.loads(path.read_text(encoding="utf-8"))
|
|
if not isinstance(value, dict):
|
|
raise ValueError("JSON value is not an object")
|
|
return value
|
|
|
|
def _validate_identity_email_delivery(values: dict[str, str]) -> Check:
|
|
mode = values.get("CONTRACT_API_IDENTITY_EMAIL_DELIVERY_MODE", "").strip()
|
|
self_service = values.get("CONTRACT_API_IDENTITY_SELF_SERVICE_ENABLED", "").strip().lower()
|
|
manual_invitation = values.get("CONTRACT_API_IDENTITY_MANUAL_INVITATION_ENABLED", "").strip().lower()
|
|
if mode == "disabled":
|
|
valid = self_service == "false" and manual_invitation == "true"
|
|
return _result(
|
|
"identity_email_delivery",
|
|
valid,
|
|
"Administrator-managed disabled identity email mode is configured."
|
|
if valid
|
|
else "Disabled identity email mode requires self-service=false and manual invitations=true.",
|
|
)
|
|
if mode != "smtp":
|
|
return _result(
|
|
"identity_email_delivery",
|
|
False,
|
|
"Production identity email delivery must use disabled or smtp mode.",
|
|
)
|
|
|
|
raw_username = values.get("CONTRACT_API_IDENTITY_SMTP_USERNAME", "")
|
|
username = raw_username.strip()
|
|
password_file = values.get("CONTRACT_API_IDENTITY_SMTP_PASSWORD_FILE", "").strip()
|
|
password_file_host = values.get("CONTRACT_API_IDENTITY_SMTP_PASSWORD_FILE_HOST", "").strip()
|
|
direct_password = values.get("CONTRACT_API_IDENTITY_SMTP_PASSWORD", "")
|
|
raw_starttls = values.get("CONTRACT_API_IDENTITY_SMTP_USE_STARTTLS", "").strip().lower()
|
|
raw_use_ssl = values.get("CONTRACT_API_IDENTITY_SMTP_USE_SSL", "").strip().lower()
|
|
starttls = raw_starttls == "true"
|
|
use_ssl = raw_use_ssl == "true"
|
|
raw_public_url = values.get("CONTRACT_API_IDENTITY_PUBLIC_WEB_BASE_URL", "")
|
|
public_url = urlparse(raw_public_url)
|
|
raw_smtp_host = values.get("CONTRACT_API_IDENTITY_SMTP_HOST", "")
|
|
smtp_port = _parse_int(values.get("CONTRACT_API_IDENTITY_SMTP_PORT", ""))
|
|
smtp_timeout = _parse_float(
|
|
values.get("CONTRACT_API_IDENTITY_SMTP_TIMEOUT_SECONDS", "")
|
|
)
|
|
invalid: list[str] = []
|
|
if self_service not in {"true", "false"}:
|
|
invalid.append("CONTRACT_API_IDENTITY_SELF_SERVICE_ENABLED")
|
|
if manual_invitation != "false":
|
|
invalid.append("CONTRACT_API_IDENTITY_MANUAL_INVITATION_ENABLED")
|
|
if (
|
|
not raw_smtp_host
|
|
or raw_smtp_host != raw_smtp_host.strip()
|
|
or any(character.isspace() for character in raw_smtp_host)
|
|
or _contains_control_character(raw_smtp_host)
|
|
):
|
|
invalid.append("CONTRACT_API_IDENTITY_SMTP_HOST")
|
|
if smtp_port is None or not 1 <= smtp_port <= 65535:
|
|
invalid.append("CONTRACT_API_IDENTITY_SMTP_PORT")
|
|
if smtp_timeout is None or not 0 < smtp_timeout <= 120:
|
|
invalid.append("CONTRACT_API_IDENTITY_SMTP_TIMEOUT_SECONDS")
|
|
if not _is_valid_email_address(
|
|
values.get("CONTRACT_API_IDENTITY_SMTP_FROM_EMAIL", "")
|
|
):
|
|
invalid.append("CONTRACT_API_IDENTITY_SMTP_FROM_EMAIL")
|
|
if _contains_control_character(
|
|
values.get("CONTRACT_API_IDENTITY_SMTP_FROM_NAME", "")
|
|
):
|
|
invalid.append("CONTRACT_API_IDENTITY_SMTP_FROM_NAME")
|
|
if raw_username != username or _contains_control_character(raw_username):
|
|
invalid.append("CONTRACT_API_IDENTITY_SMTP_USERNAME")
|
|
if raw_starttls not in {"true", "false"}:
|
|
invalid.append("CONTRACT_API_IDENTITY_SMTP_USE_STARTTLS")
|
|
if raw_use_ssl not in {"true", "false"}:
|
|
invalid.append("CONTRACT_API_IDENTITY_SMTP_USE_SSL")
|
|
if starttls == use_ssl:
|
|
invalid.append("CONTRACT_API_IDENTITY_SMTP_TLS_MODE")
|
|
if direct_password:
|
|
invalid.append("CONTRACT_API_IDENTITY_SMTP_PASSWORD")
|
|
if username:
|
|
if password_file != SMTP_CREDENTIAL_CONTAINER_PATH:
|
|
invalid.append("CONTRACT_API_IDENTITY_SMTP_PASSWORD_FILE")
|
|
if not password_file_host or not Path(password_file_host).is_absolute():
|
|
invalid.append("CONTRACT_API_IDENTITY_SMTP_PASSWORD_FILE_HOST")
|
|
elif password_file or password_file_host:
|
|
invalid.append("CONTRACT_API_IDENTITY_SMTP_USERNAME")
|
|
if (
|
|
raw_public_url != raw_public_url.strip()
|
|
or _contains_control_character(raw_public_url)
|
|
or public_url.scheme != "https"
|
|
or not public_url.netloc
|
|
or public_url.hostname in LOOPBACK_VALUES
|
|
):
|
|
invalid.append("CONTRACT_API_IDENTITY_PUBLIC_WEB_BASE_URL")
|
|
valid = not invalid
|
|
return _result(
|
|
"identity_email_delivery",
|
|
valid,
|
|
"SMTP identity email mode uses an encrypted transport and protected password file."
|
|
if valid
|
|
else f"Invalid SMTP identity email keys: {', '.join(sorted(set(invalid)))}.",
|
|
)
|
|
|
|
|
|
def _parse_int(value: str) -> int | None:
|
|
try:
|
|
return int(value)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def _validate_ai_timeout_budget(values: dict[str, str]) -> Check:
|
|
contract_timeout = _parse_float(values.get("CONTRACT_API_AI_PLATFORM_TIMEOUT_SECONDS", ""))
|
|
provider_timeout = _parse_float(values.get("AI_PLATFORM_PROVIDER_REQUEST_TIMEOUT_SECONDS", ""))
|
|
valid = (
|
|
contract_timeout is not None
|
|
and provider_timeout is not None
|
|
and math.isfinite(contract_timeout)
|
|
and math.isfinite(provider_timeout)
|
|
and 1 <= contract_timeout <= 600
|
|
and provider_timeout > 0
|
|
and contract_timeout >= provider_timeout + 5
|
|
)
|
|
return _result(
|
|
"ai_timeout_budget",
|
|
valid,
|
|
"Contract-to-AI timeout leaves at least five seconds beyond the Provider request budget."
|
|
if valid
|
|
else "Timeouts must be finite and positive, Contract must be within 1..600 seconds, and Contract must exceed Provider by at least five seconds.",
|
|
)
|
|
|
|
|
|
def _validate_capacity_thresholds(values: dict[str, str]) -> Check:
|
|
prefixes = ("ROOT_DISK", "HOME_DISK", "BACKUP_DISK")
|
|
invalid: list[str] = []
|
|
for prefix in prefixes:
|
|
warning_percent = _parse_float(values.get(f"{prefix}_WARNING_PERCENT", ""))
|
|
critical_percent = _parse_float(values.get(f"{prefix}_CRITICAL_PERCENT", ""))
|
|
warning_free_gib = _parse_float(values.get(f"{prefix}_WARNING_FREE_GIB", ""))
|
|
critical_free_gib = _parse_float(values.get(f"{prefix}_CRITICAL_FREE_GIB", ""))
|
|
finite = all(
|
|
value is not None and math.isfinite(value)
|
|
for value in (warning_percent, critical_percent, warning_free_gib, critical_free_gib)
|
|
)
|
|
valid = bool(
|
|
finite
|
|
and warning_percent is not None
|
|
and critical_percent is not None
|
|
and warning_free_gib is not None
|
|
and critical_free_gib is not None
|
|
and 0 <= warning_percent < critical_percent <= 100
|
|
and 0 <= critical_free_gib <= warning_free_gib
|
|
)
|
|
if not valid:
|
|
invalid.append(prefix)
|
|
return _result(
|
|
"capacity_thresholds",
|
|
not invalid,
|
|
"Root, /home, and backup capacity thresholds are finite and ordered."
|
|
if not invalid
|
|
else f"Invalid or missing capacity threshold groups: {', '.join(invalid)}.",
|
|
)
|
|
|
|
|
|
def _validate_launch_scope_exclusion(
|
|
values: dict[str, str],
|
|
*,
|
|
manifest_path: Path = LAUNCH_SCOPE_MANIFEST_PATH,
|
|
) -> list[Check]:
|
|
try:
|
|
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
|
except (OSError, UnicodeError, json.JSONDecodeError):
|
|
manifest = None
|
|
|
|
active_flags = sorted(
|
|
key
|
|
for key in FORBIDDEN_SUPPORT_ACCESS_ACTIVATION_KEYS
|
|
if values.get(key, "").strip().lower() not in {"", "0", "false", "no", "off"}
|
|
)
|
|
return [
|
|
_result(
|
|
"launch_scope_support_access_exclusion",
|
|
manifest == EXPECTED_LAUNCH_SCOPE,
|
|
"The versioned launch scope excludes support access, keeps its kill switch engaged, and keeps identity execution disabled."
|
|
if manifest == EXPECTED_LAUNCH_SCOPE
|
|
else "The versioned launch-scope manifest is missing, invalid, or does not preserve the support-access exclusion.",
|
|
),
|
|
_result(
|
|
"launch_scope_activation_flags",
|
|
not active_flags,
|
|
"No production activation flag enables support access, identity execution, or destructive identity-write APIs."
|
|
if not active_flags
|
|
else f"Forbidden support-access or identity-write activation flags are enabled: {', '.join(active_flags)}.",
|
|
),
|
|
]
|
|
|
|
|
|
def _parse_float(value: str) -> float | None:
|
|
try:
|
|
return float(value)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def _is_valid_email_address(value: str) -> bool:
|
|
if (
|
|
not value
|
|
or value != value.strip()
|
|
or any(character.isspace() for character in value)
|
|
or _contains_control_character(value)
|
|
):
|
|
return False
|
|
try:
|
|
address = Address(addr_spec=value)
|
|
except (TypeError, ValueError):
|
|
return False
|
|
return bool(address.username and address.domain)
|
|
|
|
|
|
def _contains_control_character(value: str) -> bool:
|
|
return any(ord(character) < 32 or ord(character) == 127 for character in value)
|
|
|
|
|
|
def _validate_database_urls(values: dict[str, str]) -> list[Check]:
|
|
pairs = (
|
|
("CONTRACT_API_DATABASE_URL", "CONTRACT_POSTGRES_DB"),
|
|
("AI_PLATFORM_DATABASE_URL", "AI_PLATFORM_POSTGRES_DB"),
|
|
("YUQEI_WORKER_DATABASE_URL", "AI_PLATFORM_POSTGRES_DB"),
|
|
)
|
|
problems: list[str] = []
|
|
for url_key, database_key in pairs:
|
|
raw_url = values.get(url_key, "")
|
|
parseable_url = raw_url.replace("postgresql+psycopg://", "postgresql://", 1)
|
|
parsed = urlparse(parseable_url)
|
|
database_name = parsed.path.lstrip("/")
|
|
expected_database = values.get(database_key, "")
|
|
if parsed.scheme != "postgresql" or parsed.hostname != "postgres" or database_name != expected_database:
|
|
problems.append(url_key)
|
|
if any(marker in database_name.lower() for marker in ("smoke", "test", "dev")):
|
|
problems.append(url_key)
|
|
database_names = [values.get(key, "") for key in ("CONTRACT_POSTGRES_DB", "AI_PLATFORM_POSTGRES_DB", "OPS_POSTGRES_DB")]
|
|
if any(not IDENTIFIER_PATTERN.fullmatch(name) for name in database_names if name):
|
|
problems.append("database_names")
|
|
if len([name for name in database_names if name]) != len(set(name for name in database_names if name)):
|
|
problems.append("database_names_not_unique")
|
|
return [_result("database_boundaries", not problems, "Database URLs target distinct production PostgreSQL databases on the internal service host." if not problems else f"Invalid database configuration keys: {', '.join(sorted(set(problems)))}.")]
|
|
|
|
|
|
def _validate_public_policy_deployment(
|
|
values: dict[str, str],
|
|
*,
|
|
template_mode: bool,
|
|
) -> list[Check]:
|
|
candidate_dir = values.get("PUBLIC_POLICY_CANDIDATE_DIR_HOST", "").strip()
|
|
approval_file = values.get("PUBLIC_POLICY_LEGAL_APPROVAL_FILE_HOST", "").strip()
|
|
trusted_keyring = values.get("PUBLIC_POLICY_TRUSTED_KEYRING_FILE_HOST", "").strip()
|
|
release_digest = values.get("CONTRACT_API_PUBLIC_POLICY_RELEASE_DIGEST", "").strip()
|
|
contract_image = values.get("CONTRACT_API_IMAGE", "").strip()
|
|
deployed_contract_digest = contract_image.rsplit("@", 1)[1] if "@" in contract_image else ""
|
|
|
|
paths = (candidate_dir, approval_file, trusted_keyring)
|
|
paths_ok = (
|
|
all(paths)
|
|
and all(_is_absolute_host_path(path) for path in paths)
|
|
and len(set(paths)) == len(paths)
|
|
)
|
|
executor_ok = values.get("CONTRACT_API_PUBLIC_POLICY_EXECUTOR_ENABLED", "").strip().lower() == "false"
|
|
digest_ok = (
|
|
bool(IMAGE_DIGEST_PATTERN.fullmatch(release_digest))
|
|
and release_digest == deployed_contract_digest.lower()
|
|
)
|
|
deployment_ok = paths_ok and executor_ok and digest_ok
|
|
|
|
forbidden = sorted(
|
|
key
|
|
for key, value in values.items()
|
|
if value.strip() and _is_forbidden_public_policy_key(key)
|
|
)
|
|
return [
|
|
_result(
|
|
"public_policy_deployment",
|
|
template_mode or deployment_ok,
|
|
"Public policy evidence uses distinct absolute paths, the executor is disabled, and the release digest matches the Contract API image."
|
|
if deployment_ok
|
|
else "Public policy requires distinct absolute evidence paths, executor=false, and a release digest matching the immutable Contract API image.",
|
|
warning=template_mode and not deployment_ok,
|
|
),
|
|
_result(
|
|
"public_policy_credential_isolation",
|
|
not forbidden,
|
|
"Production contains no public-policy private key, password, or staging credential."
|
|
if not forbidden
|
|
else f"Forbidden public-policy credentials are configured: {', '.join(forbidden)}.",
|
|
),
|
|
]
|
|
|
|
|
|
def _is_forbidden_public_policy_key(key: str) -> bool:
|
|
normalized = key.upper()
|
|
if "PUBLIC_POLICY" not in normalized:
|
|
return False
|
|
return any(token in normalized for token in ("PRIVATE_KEY", "PRIVATE_KEY_FILE", "PASSWORD", "SECRET", "STAGING"))
|
|
|
|
def _validate_provider_acceptance_deployment(
|
|
values: dict[str, str],
|
|
*,
|
|
template_mode: bool,
|
|
) -> list[Check]:
|
|
receipt_path = values.get("AI_PLATFORM_PROVIDER_ACCEPTANCE_RECEIPT_FILE_HOST", "").strip()
|
|
trusted_keys_path = values.get(
|
|
"AI_PLATFORM_PROVIDER_ACCEPTANCE_TRUSTED_KEYS_FILE_HOST",
|
|
"",
|
|
).strip()
|
|
receipt_hash = values.get("AI_PLATFORM_PROVIDER_ACCEPTANCE_EXPECTED_RECEIPT_SHA256", "").strip()
|
|
max_ttl_seconds = _parse_int(
|
|
values.get("AI_PLATFORM_PROVIDER_ACCEPTANCE_RECEIPT_MAX_TTL_SECONDS", "")
|
|
)
|
|
clock_skew_seconds = _parse_int(
|
|
values.get("AI_PLATFORM_PROVIDER_ACCEPTANCE_RECEIPT_CLOCK_SKEW_SECONDS", "")
|
|
)
|
|
paths_ok = (
|
|
bool(receipt_path)
|
|
and bool(trusted_keys_path)
|
|
and _is_absolute_host_path(receipt_path)
|
|
and _is_absolute_host_path(trusted_keys_path)
|
|
and receipt_path != trusted_keys_path
|
|
)
|
|
hash_ok = bool(SHA256_PATTERN.fullmatch(receipt_hash)) and len(set(receipt_hash)) > 1
|
|
time_policy_ok = (
|
|
max_ttl_seconds is not None
|
|
and 300 <= max_ttl_seconds <= 604800
|
|
and clock_skew_seconds is not None
|
|
and 0 <= clock_skew_seconds <= 300
|
|
)
|
|
deployment_ok = paths_ok and hash_ok and time_policy_ok
|
|
|
|
forbidden = sorted(
|
|
key
|
|
for key, value in values.items()
|
|
if value.strip() and _is_forbidden_production_promotion_key(key)
|
|
)
|
|
return [
|
|
_result(
|
|
"provider_acceptance_deployment",
|
|
template_mode or deployment_ok,
|
|
"Provider acceptance uses distinct absolute receipt/keyring paths and a pinned receipt SHA-256."
|
|
if deployment_ok
|
|
else "Provider acceptance requires distinct absolute receipt/keyring paths, a non-placeholder lowercase SHA-256 pin, a 300-604800 second TTL, and at most 300 seconds of clock skew.",
|
|
warning=template_mode and not deployment_ok,
|
|
),
|
|
_result(
|
|
"provider_acceptance_credential_isolation",
|
|
not forbidden,
|
|
"Production contains no Provider acceptance signing private key, legal attestation secret, or staging database credential."
|
|
if not forbidden
|
|
else f"Forbidden production Provider acceptance credentials are configured: {', '.join(forbidden)}.",
|
|
),
|
|
]
|
|
|
|
|
|
def _is_forbidden_production_promotion_key(key: str) -> bool:
|
|
normalized = key.upper()
|
|
if normalized == "AI_PLATFORM_PROVIDER_LEGAL_ATTESTATION_SECRET":
|
|
return True
|
|
if "PROVIDER_ACCEPTANCE" in normalized and (
|
|
"PRIVATE_KEY" in normalized or "SIGNING_KEY" in normalized
|
|
):
|
|
return True
|
|
has_staging_scope = bool(re.search(r"(^|_)STAGING($|_)", normalized))
|
|
has_database_scope = bool(re.search(r"(^|_)(DATABASE|DB|POSTGRES)($|_)", normalized))
|
|
return has_staging_scope and has_database_scope
|
|
|
|
|
|
def _is_bounded_absolute_file(path: Path, max_bytes: int) -> bool:
|
|
try:
|
|
return path.is_absolute() and path.is_file() and 0 < path.stat().st_size <= max_bytes
|
|
except OSError:
|
|
return False
|
|
|
|
|
|
def _is_absolute_host_path(value: str) -> bool:
|
|
return Path(value).is_absolute() or value.startswith("/")
|
|
|
|
|
|
def _allowed_example_value(key: str, values: dict[str, str]) -> bool:
|
|
email_disabled = values.get("CONTRACT_API_IDENTITY_EMAIL_DELIVERY_MODE") == "disabled"
|
|
if email_disabled and (
|
|
key.startswith("CONTRACT_API_IDENTITY_SMTP_")
|
|
or key == "CONTRACT_API_IDENTITY_PUBLIC_WEB_BASE_URL"
|
|
):
|
|
return True
|
|
return False
|
|
|
|
|
|
def _contains_placeholder(value: str) -> bool:
|
|
return any(marker in value.lower() for marker in PLACEHOLDER_MARKERS)
|
|
|
|
|
|
def _result(
|
|
key: str,
|
|
passed: bool,
|
|
message: str,
|
|
*,
|
|
warning: bool = False,
|
|
details: dict[str, object] | None = None,
|
|
) -> Check:
|
|
status = "warning" if warning else ("pass" if passed else "fail")
|
|
return Check(key=key, status=status, message=message, details=details or {})
|
|
|
|
|
|
def _parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--env-file", type=Path, default=REPO_ROOT / "yuqei-infra" / "compose" / ".env.production")
|
|
parser.add_argument("--artifact-dir", type=Path)
|
|
parser.add_argument("--template", action="store_true", help="Validate the committed example without requiring real secrets or files.")
|
|
return parser.parse_args()
|
|
|
|
|
|
def main() -> int:
|
|
args = _parse_args()
|
|
env_file = args.env_file.resolve()
|
|
if not env_file.is_file():
|
|
print(f"PRODUCTION_PREFLIGHT_FAIL env_file_missing={env_file}")
|
|
return 2
|
|
try:
|
|
values = parse_env_file(env_file)
|
|
except (OSError, UnicodeError, ValueError) as exc:
|
|
print(f"PRODUCTION_PREFLIGHT_FAIL env_file_invalid={type(exc).__name__}")
|
|
return 2
|
|
|
|
checks = validate_environment(values, template_mode=args.template)
|
|
checks.extend(validate_files(values, template_mode=args.template))
|
|
required_failures = [check for check in checks if check.status == "fail"]
|
|
status_value = "pass" if not required_failures else "fail"
|
|
checked_at = datetime.now(timezone.utc).astimezone()
|
|
artifact_dir = (args.artifact_dir or DEFAULT_ARTIFACT_ROOT / checked_at.strftime("%Y-%m-%dT%H-%M-%S")).resolve()
|
|
artifact_dir.mkdir(parents=True, exist_ok=True)
|
|
summary = {
|
|
"status": status_value,
|
|
"checked_at": checked_at.isoformat(),
|
|
"template_mode": args.template,
|
|
"env_file_name": env_file.name,
|
|
"checks": [asdict(check) for check in checks],
|
|
"failed_checks": [check.key for check in required_failures],
|
|
}
|
|
summary_path = artifact_dir / "production-preflight-summary.json"
|
|
summary_path.write_text(json.dumps(summary, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
print(f"PRODUCTION_PREFLIGHT_{status_value.upper()} failed={len(required_failures)} artifact={summary_path}")
|
|
return 0 if not required_failures else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|