test_environments.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 1 | """Section 7.1 — Environments: isolation, config, and secrets tests. |
| 2 | |
| 3 | Covers: |
| 4 | Local dev : docker-compose.override.yml injects DEBUG=true; bind mounts |
| 5 | keep the image separate from staging/prod. |
| 6 | Staging : separate instance name in aws-provision-staging.sh; distinct |
| 7 | domain in setup-ec2-staging.sh; same Docker image as prod. |
| 8 | Production : SSH never open to 0.0.0.0/0; startup guard rejects weak |
| 9 | DB_PASSWORD; isolated named volume for object store. |
| 10 | Config/secrets: all secrets injected via env vars; .env excluded from |
| 11 | snapshots; .env.example documents every required var; |
| 12 | no hardcoded secrets in config.py defaults. |
| 13 | """ |
| 14 | from __future__ import annotations |
| 15 | |
| 16 | import ast |
| 17 | import re |
| 18 | import subprocess |
| 19 | from pathlib import Path |
| 20 | |
| 21 | import pytest |
| 22 | |
| 23 | _ROOT = Path(__file__).resolve().parents[1] |
| 24 | |
| 25 | # ── file paths ───────────────────────────────────────────────────────────────── |
| 26 | _MUSEIGNORE = _ROOT / ".museignore" |
| 27 | _ENV_EXAMPLE = _ROOT / ".env.example" |
| 28 | _CONFIG_PY = _ROOT / "musehub" / "config.py" |
| 29 | _MAIN_PY = _ROOT / "musehub" / "main.py" |
| 30 | _COMPOSE = _ROOT / "docker-compose.yml" |
| 31 | _COMPOSE_OVER = _ROOT / "docker-compose.override.yml" |
| 32 | _PROVISION = _ROOT / "deploy" / "aws-provision.sh" |
| 33 | _PROVISION_STG = _ROOT / "deploy" / "aws-provision-staging.sh" |
| 34 | _SETUP_EC2 = _ROOT / "deploy" / "setup-ec2.sh" |
| 35 | _SETUP_STG = _ROOT / "deploy" / "setup-ec2-staging.sh" |
| 36 | |
| 37 | |
| 38 | # ═══════════════════════════════════════════════════════════════════════════════ |
| 39 | # Secrets never committed |
| 40 | # ═══════════════════════════════════════════════════════════════════════════════ |
| 41 | |
| 42 | from musehub.config import Settings |
| 43 | from pydantic_settings import SettingsConfigDict |
| 44 | |
| 45 | |
| 46 | class _TestSettings(Settings): |
| 47 | """Settings subclass for tests — does not load from any .env file.""" |
| 48 | model_config = SettingsConfigDict( |
| 49 | env_file=None, |
| 50 | env_file_encoding="utf-8", |
| 51 | extra="ignore", |
| 52 | ) |
| 53 | |
| 54 | |
| 55 | class TestSecretsNotCommitted: |
| 56 | def test_museignore_excludes_env(self) -> None: |
| 57 | """.env must be listed in .museignore so it is never snapshotted.""" |
| 58 | src = _MUSEIGNORE.read_text() |
| 59 | # Accept both bare ".env" and glob patterns like ".env.*" |
| 60 | assert re.search(r'["\']\s*\.env\s*["\']', src), ( |
| 61 | ".env is not in .museignore — it would be snapshotted and committed" |
| 62 | ) |
| 63 | |
| 64 | def test_museignore_excludes_env_star(self) -> None: |
| 65 | """Explicit environment-specific .env variants must be in .museignore. |
| 66 | |
| 67 | We use explicit entries instead of a .env.* glob so that .env.example |
| 68 | (a committed template convention) is not accidentally excluded. |
| 69 | """ |
| 70 | src = _MUSEIGNORE.read_text() |
| 71 | for variant in (".env.staging", ".env.production", ".env.prod", ".env.local"): |
| 72 | assert variant in src, ( |
| 73 | f"{variant} is not in .museignore — environment-specific files could be committed" |
| 74 | ) |
| 75 | |
| 76 | def test_env_example_exists(self) -> None: |
| 77 | """.env.example must exist as a safe template without real secrets.""" |
| 78 | assert _ENV_EXAMPLE.exists(), ".env.example missing — engineers have no config template" |
| 79 | |
| 80 | def test_no_hardcoded_secrets_in_config_defaults(self) -> None: |
| 81 | """config.py defaults must not contain real credentials. |
| 82 | |
| 83 | Checks that string defaults for secret fields are None (not literal passwords). |
| 84 | Parses the AST to be precise — no false positives from comments. |
| 85 | """ |
| 86 | tree = ast.parse(_CONFIG_PY.read_text()) |
| 87 | _SECRET_FIELDS = { |
| 88 | "db_password", "blob_storage_access_key_id", "blob_storage_secret_access_key", |
| 89 | "webhook_secret_key", "mcp_token", |
| 90 | } |
| 91 | for node in ast.walk(tree): |
| 92 | if not isinstance(node, ast.AnnAssign): |
| 93 | continue |
| 94 | if not isinstance(node.target, ast.Name): |
| 95 | continue |
| 96 | fname = node.target.id |
| 97 | if fname not in _SECRET_FIELDS: |
| 98 | continue |
| 99 | # Default must be None, not a literal string |
| 100 | if node.value is not None and isinstance(node.value, ast.Constant): |
| 101 | val = node.value.value |
| 102 | if val is not None: |
| 103 | pytest.fail( |
| 104 | f"config.py: {fname!r} has a hardcoded default {val!r} — " |
| 105 | "secrets must default to None" |
| 106 | ) |
| 107 | |
| 108 | def test_env_example_has_no_real_secrets(self) -> None: |
| 109 | """.env.example must not contain any value that looks like a generated secret. |
| 110 | |
| 111 | A generated secret has ≥ 32 random hex chars or a Fernet key pattern. |
| 112 | Real values like 'musehub' or 'changeme' are expected (they're documented |
| 113 | weak values, not actual secrets). |
| 114 | """ |
| 115 | src = _ENV_EXAMPLE.read_text() |
| 116 | # Fernet key pattern: URL-safe base64, exactly 44 chars ending in '=' |
| 117 | fernet_pattern = re.compile(r'[A-Za-z0-9_\-]{43}=') |
| 118 | assert not fernet_pattern.search(src), ( |
| 119 | ".env.example appears to contain a real Fernet key" |
| 120 | ) |
| 121 | # Generated hex secret: 32+ hex chars on a value line (not a comment) |
| 122 | for line in src.splitlines(): |
| 123 | line = line.strip() |
| 124 | if line.startswith("#") or "=" not in line: |
| 125 | continue |
| 126 | _, _, value = line.partition("=") |
| 127 | value = value.strip().strip('"\'') |
| 128 | if re.fullmatch(r"[0-9a-f]{32,}", value): |
| 129 | pytest.fail( |
| 130 | f".env.example contains what looks like a real secret: {line!r}" |
| 131 | ) |
| 132 | |
| 133 | |
| 134 | # ═══════════════════════════════════════════════════════════════════════════════ |
| 135 | # .env.example completeness |
| 136 | # ═══════════════════════════════════════════════════════════════════════════════ |
| 137 | |
| 138 | class TestEnvExampleCompleteness: |
| 139 | _src = _ENV_EXAMPLE.read_text() |
| 140 | |
| 141 | def test_documents_muse_env(self) -> None: |
| 142 | assert "MUSE_ENV" in self._src |
| 143 | |
| 144 | def test_documents_db_password(self) -> None: |
| 145 | assert "DB_PASSWORD" in self._src |
| 146 | |
| 147 | def test_documents_debug(self) -> None: |
| 148 | assert "DEBUG" in self._src |
| 149 | |
| 150 | def test_documents_cors_origins(self) -> None: |
| 151 | assert "CORS_ORIGINS" in self._src |
| 152 | |
| 153 | def test_documents_webhook_secret_key(self) -> None: |
| 154 | assert "WEBHOOK_SECRET_KEY" in self._src |
| 155 | |
| 156 | def test_documents_runner_token(self) -> None: |
| 157 | assert "RUNNER_TOKEN" in self._src |
| 158 | |
| 159 | def test_warns_against_debug_in_prod(self) -> None: |
| 160 | assert re.search(r"NEVER.*DEBUG.*prod|prod.*NEVER.*DEBUG|staging.*production", self._src, re.IGNORECASE), ( |
| 161 | ".env.example should warn against DEBUG=true in staging/production" |
| 162 | ) |
| 163 | |
| 164 | def test_documents_staging_and_production_values(self) -> None: |
| 165 | """Must have guidance for at least staging and production.""" |
| 166 | assert "staging" in self._src.lower() |
| 167 | assert "production" in self._src.lower() |
| 168 | |
| 169 | def test_documents_secret_generation_commands(self) -> None: |
| 170 | """Must show how to generate secrets — engineers should never guess.""" |
| 171 | assert "openssl rand" in self._src or "Fernet.generate_key" in self._src |
| 172 | |
| 173 | |
| 174 | # ═══════════════════════════════════════════════════════════════════════════════ |
| 175 | # Local dev: docker compose isolation |
| 176 | # ═══════════════════════════════════════════════════════════════════════════════ |
| 177 | |
| 178 | class TestLocalDevIsolation: |
| 179 | def test_override_sets_debug_true(self) -> None: |
| 180 | """docker-compose.override.yml must set DEBUG=true for local dev.""" |
| 181 | src = _COMPOSE_OVER.read_text() |
| 182 | assert re.search(r'DEBUG.*true', src, re.IGNORECASE), ( |
| 183 | "override.yml does not set DEBUG=true — local dev won't have debug mode" |
| 184 | ) |
| 185 | |
| 186 | def test_main_compose_does_not_set_debug_true(self) -> None: |
| 187 | """docker-compose.yml must NOT set DEBUG=true — would be a prod footgun.""" |
| 188 | src = _COMPOSE.read_text() |
| 189 | # Allow DEBUG references in comments but not as a live env value |
| 190 | for line in src.splitlines(): |
| 191 | stripped = line.strip() |
| 192 | if stripped.startswith("#"): |
| 193 | continue |
| 194 | # Pattern: DEBUG: "true" or DEBUG=true |
| 195 | if re.search(r'DEBUG[:\s]*["\']?true["\']?', stripped, re.IGNORECASE): |
| 196 | pytest.fail( |
| 197 | f"docker-compose.yml sets DEBUG=true on line: {line!r} — " |
| 198 | "this would enable debug mode in production" |
| 199 | ) |
| 200 | |
| 201 | def test_override_has_bind_mounts(self) -> None: |
| 202 | """override.yml must bind-mount source dirs for live reload.""" |
| 203 | src = _COMPOSE_OVER.read_text() |
| 204 | assert "./musehub:/app/musehub" in src, ( |
| 205 | "override.yml should bind-mount ./musehub for live reload" |
| 206 | ) |
| 207 | |
| 208 | def test_object_store_uses_named_volume(self) -> None: |
| 209 | """musehub_data volume in docker-compose.yml isolates object store from host.""" |
| 210 | src = _COMPOSE.read_text() |
| 211 | assert "musehub_data:/data" in src, ( |
| 212 | "Object store must be mounted via a named volume — not a bind mount — " |
| 213 | "so it survives container restarts and can be backed up independently" |
| 214 | ) |
| 215 | |
| 216 | def test_postgres_uses_named_volume(self) -> None: |
| 217 | """postgres_data volume isolates DB from host filesystem.""" |
| 218 | src = _COMPOSE.read_text() |
| 219 | assert "postgres_data:/var/lib/postgresql/data" in src |
| 220 | |
| 221 | |
| 222 | # ═══════════════════════════════════════════════════════════════════════════════ |
| 223 | # Staging: separate instance and domain |
| 224 | # ═══════════════════════════════════════════════════════════════════════════════ |
| 225 | |
| 226 | class TestStagingIsolation: |
| 227 | def test_staging_instance_name_differs_from_prod(self) -> None: |
| 228 | """Staging and prod must use different EC2 instance names.""" |
| 229 | prod_src = _PROVISION.read_text() |
| 230 | stg_src = _PROVISION_STG.read_text() |
| 231 | |
| 232 | prod_name = re.search(r'INSTANCE_NAME=["\']?(\S+)["\']?', prod_src) |
| 233 | stg_name = re.search(r'INSTANCE_NAME=["\']?(\S+)["\']?', stg_src) |
| 234 | assert prod_name and stg_name, "INSTANCE_NAME not found in provision scripts" |
| 235 | assert prod_name.group(1) != stg_name.group(1), ( |
| 236 | f"Staging and prod use the same INSTANCE_NAME ({prod_name.group(1)!r}) — " |
| 237 | "they would overwrite each other" |
| 238 | ) |
| 239 | |
| 240 | def test_staging_uses_staging_domain(self) -> None: |
| 241 | """setup-ec2-staging.sh must reference a staging domain, not the prod domain.""" |
| 242 | src = _SETUP_STG.read_text() |
| 243 | assert "staging" in src.lower(), ( |
| 244 | "setup-ec2-staging.sh does not reference a staging domain" |
| 245 | ) |
| 246 | # Must not set DOMAIN to the production domain without qualification |
| 247 | domain_match = re.search(r'^DOMAIN=["\']?(\S+)["\']?', src, re.MULTILINE) |
| 248 | if domain_match: |
| 249 | domain = domain_match.group(1) |
| 250 | assert "staging" in domain.lower(), ( |
| 251 | f"setup-ec2-staging.sh DOMAIN={domain!r} does not look like a staging domain" |
| 252 | ) |
| 253 | |
| 254 | def test_same_ami_for_staging_and_prod(self) -> None: |
| 255 | """Staging and prod must use the same AMI (same Docker image = full mirror).""" |
| 256 | prod_src = _PROVISION.read_text() |
| 257 | stg_src = _PROVISION_STG.read_text() |
| 258 | prod_ami = re.search(r'AMI_ID=["\']?(\S+)["\']?', prod_src) |
| 259 | stg_ami = re.search(r'AMI_ID=["\']?(\S+)["\']?', stg_src) |
| 260 | if prod_ami and stg_ami: |
| 261 | assert prod_ami.group(1) == stg_ami.group(1), ( |
| 262 | "Staging and prod use different AMIs — staging is no longer a full mirror" |
| 263 | ) |
| 264 | |
| 265 | def test_staging_env_documented_in_env_example(self) -> None: |
| 266 | src = _ENV_EXAMPLE.read_text() |
| 267 | assert re.search(r"staging", src, re.IGNORECASE), ( |
| 268 | ".env.example has no staging guidance" |
| 269 | ) |
| 270 | |
| 271 | |
| 272 | # ═══════════════════════════════════════════════════════════════════════════════ |
| 273 | # Production: network hardening |
| 274 | # ═══════════════════════════════════════════════════════════════════════════════ |
| 275 | |
| 276 | class TestProductionNetworkHardening: |
| 277 | def test_ssh_not_open_to_all_in_provision(self) -> None: |
| 278 | """Port 22 must NOT be open to 0.0.0.0/0 in aws-provision.sh.""" |
| 279 | src = _PROVISION.read_text() |
| 280 | # Pattern: add_rule 22 "0.0.0.0/0" or similar |
| 281 | for line in src.splitlines(): |
| 282 | if re.search(r'22.*0\.0\.0\.0/0|0\.0\.0\.0/0.*22', line): |
| 283 | # Allow if the line is a comment |
| 284 | stripped = line.strip() |
| 285 | if not stripped.startswith("#"): |
| 286 | pytest.fail( |
| 287 | f"Port 22 is open to 0.0.0.0/0 in aws-provision.sh: {line!r}\n" |
| 288 | "SSH should be restricted to a specific IP or bastion." |
| 289 | ) |
| 290 | |
| 291 | def test_http_redirect_in_nginx_config(self) -> None: |
| 292 | """nginx config must redirect HTTP to HTTPS (80 → 443).""" |
| 293 | nginx_conf = _ROOT / "deploy" / "nginx-cf.conf" |
| 294 | src = nginx_conf.read_text() |
| 295 | assert re.search(r'listen\s+80', src), "nginx config has no port-80 listener" |
| 296 | assert "return 301" in src or "return 308" in src, ( |
| 297 | "nginx config does not redirect port 80 to HTTPS" |
| 298 | ) |
| 299 | |
| 300 | def test_ssl_configured_in_nginx(self) -> None: |
| 301 | """nginx must be configured with SSL (port 443).""" |
| 302 | nginx_conf = _ROOT / "deploy" / "nginx-cf.conf" |
| 303 | src = nginx_conf.read_text() |
| 304 | assert re.search(r'listen\s+443\s+ssl', src) |
| 305 | assert "ssl_certificate" in src |
| 306 | |
| 307 | def test_entrypoint_uses_proxy_headers(self) -> None: |
| 308 | """uvicorn must run with --proxy-headers so Cloudflare IPs are trusted.""" |
| 309 | src = (_ROOT / "entrypoint.sh").read_text() |
| 310 | assert "--proxy-headers" in src |
| 311 | |
| 312 | |
| 313 | # ═══════════════════════════════════════════════════════════════════════════════ |
| 314 | # Settings: env-var injection and startup guards |
| 315 | # ═══════════════════════════════════════════════════════════════════════════════ |
| 316 | |
| 317 | class TestSettingsEnvInjection: |
| 318 | def test_settings_reads_from_env_file(self) -> None: |
| 319 | """Settings must be configured to read .env (pydantic-settings env_file).""" |
| 320 | src = _CONFIG_PY.read_text() |
| 321 | assert "env_file" in src and ".env" in src |
| 322 | |
| 323 | def test_muse_env_defaults_to_production(self) -> None: |
| 324 | """Default must be 'production' — fail-closed (not development by accident).""" |
| 325 | from musehub.config import Settings |
| 326 | # Instantiate without any env vars — only pydantic defaults |
| 327 | import os |
| 328 | saved = os.environ.pop("MUSE_ENV", None) |
| 329 | try: |
| 330 | s = _TestSettings() |
| 331 | assert s.muse_env == "production" |
| 332 | finally: |
| 333 | if saved is not None: |
| 334 | os.environ["MUSE_ENV"] = saved |
| 335 | |
| 336 | def test_debug_defaults_to_false(self) -> None: |
| 337 | """DEBUG must default to False — never accidentally expose Swagger.""" |
| 338 | import os |
| 339 | saved = os.environ.pop("DEBUG", None) |
| 340 | try: |
| 341 | from musehub.config import Settings |
| 342 | s = _TestSettings() |
| 343 | assert s.debug is False |
| 344 | finally: |
| 345 | if saved is not None: |
| 346 | os.environ["DEBUG"] = saved |
| 347 | |
| 348 | def test_weak_db_password_rejected_in_production(self) -> None: |
| 349 | """Production startup must refuse to start with a weak DB password.""" |
| 350 | src = _MAIN_PY.read_text() |
| 351 | # The guard checks for known weak passwords |
| 352 | assert "changeme" in src or "weak" in src.lower(), ( |
| 353 | "lifespan() does not appear to guard against weak DB passwords" |
| 354 | ) |
| 355 | assert "RuntimeError" in src or "raise" in src |
| 356 | |
| 357 | def test_production_secrets_validator_warns_on_missing_webhook_key(self, caplog: pytest.LogCaptureFixture) -> None: |
| 358 | """In production mode, missing WEBHOOK_SECRET_KEY must produce a log warning.""" |
| 359 | import logging |
| 360 | from musehub.config import Settings |
| 361 | with caplog.at_level(logging.WARNING, logger="musehub.config"): |
| 362 | _TestSettings( |
| 363 | debug=False, |
| 364 | muse_env="production", |
| 365 | db_password="strong-password-that-passes-guard", |
| 366 | webhook_secret_key=None, |
| 367 | ) |
| 368 | warning_text = " ".join(caplog.messages) |
| 369 | assert "WEBHOOK_SECRET_KEY" in warning_text or "webhook" in warning_text.lower() |
| 370 | |
| 371 | def test_production_secrets_validator_silent_in_dev(self, caplog: pytest.LogCaptureFixture) -> None: |
| 372 | """In development mode, missing secrets must NOT produce warnings.""" |
| 373 | import logging |
| 374 | from musehub.config import Settings |
| 375 | caplog.clear() |
| 376 | with caplog.at_level(logging.WARNING, logger="musehub.config"): |
| 377 | _TestSettings( |
| 378 | debug=True, |
| 379 | muse_env="development", |
| 380 | webhook_secret_key=None, |
| 381 | ) |
| 382 | assert not any("WEBHOOK_SECRET_KEY" in m for m in caplog.messages) |
| 383 | |
| 384 | def test_cors_wildcard_warns_in_non_debug(self, caplog: pytest.LogCaptureFixture) -> None: |
| 385 | """CORS=* in non-debug mode must trigger a security warning.""" |
| 386 | import logging |
| 387 | from musehub.config import Settings |
| 388 | with caplog.at_level(logging.WARNING, logger="musehub.config"): |
| 389 | _TestSettings( |
| 390 | debug=False, |
| 391 | cors_origins=["*"], |
| 392 | muse_env="test", |
| 393 | ) |
| 394 | assert any("CORS" in m or "origin" in m.lower() for m in caplog.messages) |
| 395 | |
| 396 | def test_settings_extra_env_vars_ignored(self) -> None: |
| 397 | """Unknown env vars must not cause validation errors (extra='ignore').""" |
| 398 | import os |
| 399 | os.environ["COMPLETELY_UNKNOWN_VAR"] = "value" |
| 400 | try: |
| 401 | from musehub.config import Settings |
| 402 | s = _TestSettings() |
| 403 | # Should not raise |
| 404 | finally: |
| 405 | del os.environ["COMPLETELY_UNKNOWN_VAR"] |
| 406 | |
| 407 | |
| 408 | # ═══════════════════════════════════════════════════════════════════════════════ |
| 409 | # OpenAPI exposure: disabled in production |
| 410 | # ═══════════════════════════════════════════════════════════════════════════════ |
| 411 | |
| 412 | class TestOpenAPIExposure: |
| 413 | def test_openapi_disabled_in_production(self) -> None: |
| 414 | """OpenAPI URL must be None in production (DEBUG=false, MUSE_ENV != test).""" |
| 415 | src = _MAIN_PY.read_text() |
| 416 | # The app init must conditional on debug or test env |
| 417 | assert re.search(r'openapi_url.*debug.*test|openapi_url.*None', src), ( |
| 418 | "OpenAPI endpoint does not appear to be gated by debug/test mode" |
| 419 | ) |
| 420 | |
| 421 | def test_openapi_enabled_in_test_env(self) -> None: |
| 422 | """OpenAPI must be reachable in the test environment (tests use it).""" |
| 423 | from musehub.main import app |
| 424 | from musehub.config import settings |
| 425 | if settings.muse_env == "test": |
| 426 | assert app.openapi_url is not None, ( |
| 427 | "OpenAPI URL is None in test env — tests that hit /api/openapi.json will fail" |
| 428 | ) |