import subprocess from pathlib import Path from types import SimpleNamespace import pytest from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from flask import Flask from app.config.config import ProductionConfig, apply_runtime_env_config from app.config.database_urls import ( load_and_validate_database_env, validate_database_environment, validate_postgresql_url, ) from app.core.edge_gateway.runtime_roles import provision_runtime_login from app.core.edge_gateway.service import ( EdgeGatewayConfigurationError, EdgeGatewayService, ) ROOT = Path(__file__).resolve().parents[1] PRODUCTION_ENV = { "DATABASE_URL": "postgresql://dataops_app:runtime-secret@db.example.test/dataops", "BI_AI_CATALOG_CONTROL_DATABASE_URL": "postgresql://dataops_bi_ai_catalog_control:control-secret@db.example.test/dataops", "NEO4J_URI": "bolt+s://neo4j.example.test:7687", "NEO4J_HTTP_URI": "https://neo4j.example.test:7473", "NEO4J_USER": "dataops_graph", "NEO4J_PASSWORD": "graph-secret", "MINIO_HOST": "objects.example.test:9000", "MINIO_USER": "dataops_objects", "MINIO_PASSWORD": "object-secret", "MINIO_BUCKET": "dataops-production", } def _production_app(monkeypatch, overrides=None): monkeypatch.setenv("FLASK_ENV", "production") monkeypatch.setenv("APP_ENV_FILE", "/nonexistent/dataops-test.env") for name, value in {**PRODUCTION_ENV, **(overrides or {})}.items(): if value is None: monkeypatch.delenv(name, raising=False) else: monkeypatch.setenv(name, value) app = Flask(__name__) app.config.from_object(ProductionConfig) return app @pytest.mark.parametrize( ("name", "value"), [ ("DATABASE_URL", None), ("DATABASE_URL", "postgresql://dataops_user:replace-password@127.0.0.1/dataops"), ("BI_AI_CATALOG_CONTROL_DATABASE_URL", None), ("BI_AI_CATALOG_CONTROL_DATABASE_URL", "postgresql://dataops_bi_ai_catalog_control:replace-control-password@db.example.test/dataops"), ("NEO4J_PASSWORD", None), ("NEO4J_PASSWORD", "replace-neo4j-password"), ("MINIO_USER", None), ("MINIO_PASSWORD", "replace-minio-password"), ], ) def test_production_service_configuration_fails_closed(monkeypatch, name, value): app = _production_app(monkeypatch, {name: value}) with pytest.raises(RuntimeError, match=name): apply_runtime_env_config(app) def test_production_configuration_accepts_complete_non_placeholder_values(monkeypatch): app = _production_app(monkeypatch) apply_runtime_env_config(app) assert app.config["SQLALCHEMY_DATABASE_URI"] == PRODUCTION_ENV["DATABASE_URL"] assert app.config["NEO4J_PASSWORD"] == PRODUCTION_ENV["NEO4J_PASSWORD"] assert app.config["MINIO_PASSWORD"] == PRODUCTION_ENV["MINIO_PASSWORD"] def test_runtime_and_migrator_images_are_capability_separated(): backend = (ROOT / "deploy/docker/backend.Dockerfile").read_text() runner = (ROOT / "deploy/docker/runner.Dockerfile").read_text() compose = (ROOT / "deploy/docker/docker-compose.yml").read_text() migration = (ROOT / "migrations/versions/20260809_477_edge_gateway_control_plane.py").read_text() provisioner = (ROOT / "app/core/edge_gateway/runtime_roles.py").read_text() assert "AS migrator" in backend and "AS runtime" in backend runtime = backend.split("AS runtime", 1)[1] assert "COPY migrations/" not in runtime and "COPY alembic.ini" not in runtime assert "pip uninstall -y alembic" in runtime assert "COPY migrations/" not in runner and "COPY alembic.ini" not in runner assert "pip uninstall -y alembic" in runner assert "target: migrator" in compose and "target: runtime" in compose assert "DB_ROLE_INIT_DATABASE_URL" in compose assert "DB_ROLE_INIT_DATABASE_URL" in provisioner assert "MIGRATION_DATABASE_URL" in provisioner role_bootstrap = migration.split("CREATE OR REPLACE FUNCTION", 1)[0] assert "CREATE ROLE" not in role_bootstrap and "ALTER ROLE" not in role_bootstrap def test_enterprise_edge_mtls_proxy_is_dedicated_and_fail_closed(): compose = (ROOT / "deploy/docker/docker-compose.yml").read_text() container_nginx = (ROOT / "deploy/docker/edge-mtls-nginx.conf").read_text() host_nginx = (ROOT / "deployment/config/nginx-dataops-platform.conf").read_text() assert "edge-mtls-proxy:" in compose edge_service = compose.split("edge-mtls-proxy:", 1)[1].split( "\n lightrag-neo4j:", 1 )[0] assert "enterprise-edge" in edge_service assert "edge-control-net" in edge_service assert "ipv4_address: 172.31.0.10" in edge_service assert "EDGE_MTLS_TRUSTED_PROXY_IPS: 172.31.0.10" in compose assert "EDGE_GATEWAY_SIGNING_PRIVATE_KEY_FILE: /run/secrets/edge_gateway_signing_private_key" in compose assert "EDGE_GATEWAY_SIGNING_PRIVATE_KEY:" not in compose assert "edge_gateway_signing_private_key:" in compose assert "/run/secrets/edge_gateway_signing_private_key" in compose assert "ssl_verify_client on;" in container_nginx assert "ssl_client_certificate /etc/nginx/edge-mtls/client-ca.pem;" in container_nginx assert "ssl_crl /etc/nginx/edge-mtls/client-ca.crl;" in container_nginx assert "location /api/datasource/edge/" in container_nginx assert ( "proxy_set_header X-DataOps-Edge-Client-Cert $ssl_client_escaped_cert;" in container_nginx ) assert ( "proxy_set_header X-DataOps-Edge-Client-Verify $ssl_client_verify;" in container_nginx ) assert "listen 18443 ssl;" in host_nginx assert "ssl_verify_client on;" in host_nginx assert "edge-client-ca.crt" in host_nginx assert "edge-client-ca.crl" in host_nginx def test_production_edge_signer_uses_secure_file_or_provider_and_rejects_inline(tmp_path): private_key = Ed25519PrivateKey.generate() private_path = tmp_path / "edge-signing.key" private_path.write_bytes( private_key.private_bytes( serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, serialization.NoEncryption(), ) ) private_path.chmod(0o600) public_key = private_key.public_key().public_bytes( serialization.Encoding.Raw, serialization.PublicFormat.Raw, ).hex() repository = SimpleNamespace(rollback=lambda: None) service = EdgeGatewayService( repository, signing_private_key_file=str(private_path), signing_public_key=public_key, signing_key_id="edge-key-2026-08", production=True, ) assert service._require_signer()[1] == "edge-key-2026-08" provider_service = EdgeGatewayService( repository, signing_key_provider=lambda: private_path.read_bytes(), signing_public_key=public_key, signing_key_id="edge-key-2026-08", production=True, ) assert provider_service._require_signer()[0].public_key().public_bytes( serialization.Encoding.Raw, serialization.PublicFormat.Raw, ).hex() == public_key with pytest.raises(EdgeGatewayConfigurationError, match="inline"): EdgeGatewayService( repository, signing_private_key=private_key, signing_public_key=public_key, signing_key_id="edge-key-2026-08", production=True, ) private_path.chmod(0o640) with pytest.raises(EdgeGatewayConfigurationError, match="permissions"): EdgeGatewayService( repository, signing_private_key_file=str(private_path), signing_public_key=public_key, signing_key_id="edge-key-2026-08", production=True, ) private_path.chmod(0o600) with pytest.raises(EdgeGatewayConfigurationError, match="public key"): EdgeGatewayService( repository, signing_private_key_file=str(private_path), signing_public_key="0" * 64, signing_key_id="edge-key-2026-08", production=True, ) def _template_env() -> dict[str, str]: values = {} for raw_line in (ROOT / "deployment/dataops.env").read_text(encoding="utf-8-sig").splitlines(): line = raw_line.strip() if line and not line.startswith("#") and "=" in line: name, value = line.split("=", 1) values[name] = value return values def test_checked_in_production_template_is_rejected_everywhere(monkeypatch): template = _template_env() app = _production_app(monkeypatch, template) with pytest.raises(RuntimeError, match="DATABASE_URL"): apply_runtime_env_config(app) for name, value in template.items(): monkeypatch.setenv(name, value) with pytest.raises(RuntimeError, match="DB_ROLE_INIT_DATABASE_URL"): provision_runtime_login() with pytest.raises(RuntimeError, match="DB_ROLE_INIT_DATABASE_URL"): load_and_validate_database_env(ROOT / "deployment/dataops.env") result = subprocess.run( [ ".venv/bin/python", "app/config/database_urls.py", "--env-file", "deployment/dataops.env", ], cwd=ROOT, capture_output=True, text=True, ) assert result.returncode != 0 assert "DB_ROLE_INIT_DATABASE_URL" in result.stderr deploy = (ROOT / "deployment/deploy_dataops.sh").read_text() source_deploy = (ROOT / "scripts/deploy_dataops.sh").read_text() assert "database_urls.py" in deploy and "database_urls.py" in source_deploy @pytest.mark.parametrize( ("name", "bad_value"), [ ("DB_ROLE_INIT_DATABASE_URL", "postgresql://role_admin:replace-role-init-password@db/dataops"), ("MIGRATION_DATABASE_URL", "postgresql://migrator:replace-migration-password@db/dataops"), ("DATABASE_URL", "postgresql://dataops_app:replace-runtime-password@db/dataops"), ("DATAOPS_RUNTIME_PASSWORD", "replace-runtime-password"), ], ) def test_role_init_rejects_each_database_template_secret(monkeypatch, name, bad_value): valid = { "DB_ROLE_INIT_DATABASE_URL": "postgresql://role_admin:Strong%40Role9@db/dataops", "MIGRATION_DATABASE_URL": "postgresql://migrator:Strong%40Migration9@db/dataops", "DATABASE_URL": "postgresql://dataops_app:Strong%40Runtime9@db/dataops", "BI_AI_CATALOG_CONTROL_DATABASE_URL": "postgresql://dataops_bi_ai_catalog_control:Strong%40Control9@db/dataops", "DATAOPS_MIGRATOR_USER": "dataops_migrator", "DATAOPS_RUNTIME_USER": "dataops_app", "DATAOPS_RUNTIME_PASSWORD": "Strong@Runtime9", "DATAOPS_BI_AI_CATALOG_CONTROL_USER": "dataops_bi_ai_catalog_control", "DATAOPS_BI_AI_CATALOG_CONTROL_PASSWORD": "Strong@Control9", } valid[name] = bad_value for key, value in valid.items(): monkeypatch.setenv(key, value) with pytest.raises(RuntimeError, match=name): provision_runtime_login() def test_database_url_validation_decodes_components_without_substring_false_positive(tmp_path): validate_postgresql_url( "postgresql://dataops_app:S7replace%40middle%21@db.internal/dataops", "DATABASE_URL", ) with pytest.raises(RuntimeError, match="DATABASE_URL"): validate_postgresql_url( "postgresql://dataops_app:replace%2Druntime%2Dpassword@db.internal/dataops", "DATABASE_URL", ) env_file = tmp_path / "dataops.env" env_file.write_text( "\n".join([ "DB_ROLE_INIT_DATABASE_URL=postgresql://role_admin:Strong%40Role9@db.internal/dataops", "MIGRATION_DATABASE_URL=postgresql://migrator:Strong%40Migration9@db.internal/dataops", "DATABASE_URL=postgresql://dataops_app:Strong%40Runtime9@db.internal/dataops", "DATAOPS_RUNTIME_PASSWORD=Strong@Runtime9", ]), ) load_and_validate_database_env(env_file) @pytest.mark.parametrize( "bad_url", [ "postgresql://:Strong%40Pass9@db.internal/dataops", "postgresql://replace-user:Strong%40Pass9@db.internal/dataops", "postgresql://dataops_app:@db.internal/dataops", "postgresql://dataops_app:Strong%40Pass9@/dataops", "postgresql://dataops_app:Strong%40Pass9@replace-host/dataops", "postgresql://dataops_app:Strong%40Pass9@db.internal/", "postgresql://dataops_app:Strong%40Pass9@db.internal/database_name", "postgresql://dataops_app:replace%2Dpassword@db.internal/dataops", ], ) def test_database_url_validation_rejects_each_empty_or_template_component(bad_url): with pytest.raises(RuntimeError, match="DATABASE_URL"): validate_postgresql_url(bad_url, "DATABASE_URL") def test_shared_environment_validation_compares_decoded_runtime_password(): values = { "DB_ROLE_INIT_DATABASE_URL": "postgresql://role_admin:Strong%40Role9@db.internal/dataops", "MIGRATION_DATABASE_URL": "postgresql://migrator:Strong%40Migration9@db.internal/dataops", "DATABASE_URL": "postgresql://dataops_app:Strong%40Runtime9@db.internal/dataops", "DATAOPS_RUNTIME_PASSWORD": "Strong@Runtime9", } validate_database_environment(values) values["DATAOPS_RUNTIME_PASSWORD"] = "Different@Runtime9" with pytest.raises(RuntimeError, match="DATAOPS_RUNTIME_PASSWORD.*match"): validate_database_environment(values)