knowledge-adapter-service/tests/test_delivery_artifacts.py

import json
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]


def test_dockerfile_is_reproducible_non_root_and_secret_free() -> None:
    dockerfile = (ROOT / "Dockerfile").read_text(encoding="utf-8")

    assert "python:3.12.11-slim-bookworm" in dockerfile
    assert "uv sync --frozen --no-dev --no-editable" in dockerfile
    assert "USER 10001:10001" in dockerfile
    assert "knowledge_adapter.entrypoint:create_production_app" in dockerfile
    assert "--factory" in dockerfile
    assert "--no-access-log" in dockerfile
    assert "--host\", \"0.0.0.0\"" in dockerfile
    assert "--port\", \"8080\"" in dockerfile
    for forbidden in (
        "WEKNORA_API_KEY=",
        "OSS_ACCESS_KEY_ID",
        "OSS_ACCESS_KEY_SECRET",
        "ALIBABA_CLOUD_ACCESS_KEY",
    ):
        assert forbidden not in dockerfile


def test_compose_is_hardened_and_mounts_only_secret_files() -> None:
    compose = (ROOT / "deploy/docker-compose.adapter.yml").read_text(encoding="utf-8")

    for required in (
        '"127.0.0.1:18080:8080"',
        'user: "10001:10001"',
        "read_only: true",
        "no-new-privileges:true",
        "cap_drop:",
        "- ALL",
        "pids_limit: 256",
        "tmpfs:",
        "healthcheck:",
        "WEKNORA_API_KEY_FILE: /run/secrets/weknora_api_key",
        "ASSET_DATABASE_URL_FILE: /run/secrets/asset_database_url",
        "WEKNORA_READONLY_DATABASE_URL_FILE: /run/secrets/weknora_readonly_database_url",
        'OSS_CREDENTIAL_MODE: "${OSS_CREDENTIAL_MODE:-static_ak}"',
        "OSS_ACCESS_KEY_ID_FILE: /run/secrets/oss_access_key_id",
        "OSS_ACCESS_KEY_SECRET_FILE: /run/secrets/oss_access_key_secret",
        "- adapter",
        "- weknora_upstream",
        "weknora_upstream:",
        "external: true",
        'name: "${WEKNORA_NETWORK_NAME:?set WEKNORA_NETWORK_NAME}"',
        "target: /run/secrets/weknora_api_key",
        "target: /run/secrets/asset_database_url",
        "target: /run/secrets/weknora_readonly_database_url",
        "target: /run/secrets/oss_access_key_id",
        "target: /run/secrets/oss_access_key_secret",
        "read_only: true",
    ):
        assert required in compose
    for forbidden in (
        "WEKNORA_API_KEY:",
        "OSS_ACCESS_KEY_ID:",
        "OSS_ACCESS_KEY_SECRET:",
        "ALIBABA_CLOUD_ACCESS_KEY",
        "/var/run/docker.sock",
        "postgres:",
        "redis:",
    ):
        assert forbidden not in compose


def test_nginx_config_enforces_tls_methods_limits_and_safe_logging() -> None:
    nginx = (ROOT / "deploy/nginx/knowledge-adapter.conf").read_text(encoding="utf-8")

    for required in (
        "listen 443 ssl;",
        "server_name assets.example.com;",
        "ssl_protocols TLSv1.2 TLSv1.3;",
        "access_log off;",
        "proxy_pass http://127.0.0.1:18080;",
        "proxy_set_header X-Request-ID $adapter_request_id;",
        "proxy_intercept_errors off;",
        "client_max_body_size 64k;",
        "limit_req zone=adapter_retrieval",
        "location = /retrieval",
        "limit_except POST",
        "location ~ ^/i/",
        "limit_except GET",
        "location = /healthz",
        "return 404;",
    ):
        assert required in nginx
    for forbidden in (
        "$request_uri",
        "$args",
        "$upstream_http_location",
        "proxy_set_header X-OSS",
    ):
        assert forbidden not in nginx


def test_ram_policy_is_get_only_and_prefix_scoped() -> None:
    policy_path = ROOT / "deploy/ram/knowledge-adapter-oss-read-policy.json"
    policy = json.loads(policy_path.read_text(encoding="utf-8"))

    assert policy["Version"] == "1"
    assert len(policy["Statement"]) == 1
    statement = policy["Statement"][0]
    assert statement == {
        "Effect": "Allow",
        "Action": ["oss:GetObject"],
        "Resource": [
            "acs:oss:*:*:${OSS_BUCKET}/weknora/${WEKNORA_TENANT_ID}/*"
        ],
    }
    serialized = json.dumps(policy, sort_keys=True)
    for forbidden in (
        "oss:List",
        "oss:Put",
        "oss:Delete",
        "kms:Decrypt",
    ):
        assert forbidden not in serialized


def test_delivery_docs_and_makefile_cover_required_review_gates() -> None:
    architecture = (ROOT / "docs/architecture.md").read_text(encoding="utf-8")
    security = (ROOT / "docs/security.md").read_text(encoding="utf-8")
    deployment = (ROOT / "docs/deployment.md").read_text(encoding="utf-8")
    rotation = (ROOT / "docs/key-rotation.md").read_text(encoding="utf-8")
    makefile = (ROOT / "Makefile").read_text(encoding="utf-8")

    for required in (
        "POST /retrieval",
        "GET /i/{UUID}",
        "GET /healthz",
        "WeKnora",
        "asset_mapping",
        "Private OSS",
        "static AK Secret files",
        "600",
    ):
        assert required in architecture
    for required in (
        "static AK",
        "oss:GetObject",
        "Private",
        "Presigned URL",
        "0400",
        "Authorization",
        "object_key",
        "SSRF",
        "## Sources",
    ):
        assert required in security
    for required in (
        "TEST ENVIRONMENT ONLY",
        "Production deployment is prohibited",
        "docker compose config",
        "migration",
        "rollback",
        "0400",
        "10001",
        "healthz",
        "OSS_ACCESS_KEY_ID_HOST_FILE",
        "OSS_ACCESS_KEY_SECRET_HOST_FILE",
    ):
        assert required in deployment
    for required in (
        "DIFY_KEY_HASH_CURRENT",
        "DIFY_KEY_HASH_NEXT",
        "WeKnora API Key",
        "Adapter DB",
        "WeKnora read-only DB",
        "OSS static AK",
        "rollback",
    ):
        assert required in rotation
    for required in (
        "test-slice8:",
        "lint:",
        "compose-config:",
        "check:",
    ):
        assert required in makefile
    for forbidden in ("sudo", " down -v", " compose up"):
        assert forbidden not in makefile


def test_git_and_docker_context_ignore_secret_material() -> None:
    gitignore = (ROOT / ".gitignore").read_text(encoding="utf-8").splitlines()
    dockerignore = (ROOT / ".dockerignore").read_text(encoding="utf-8").splitlines()

    for pattern in (".env", "secrets/", "*.secret", "*.key", "*.pem"):
        assert pattern in gitignore
        assert pattern in dockerignore


def test_slice9_audit_report_distinguishes_pass_blocked_and_not_run() -> None:
    report = (ROOT / "docs/audit-report-slice9.md").read_text(encoding="utf-8")

    for required in (
        "133 passed",
        "All checks passed!",
        "WeKnora /health",
        "PASS",
        "BLOCKED",
        "NOT RUN",
        "Docker daemon permission",
        "nginx binary unavailable",
        "ECS RAM Role",
        "Dify → Adapter → WeKnora → Adapter DB → OSS",
        "Private OSS anonymous denial",
        "600-second expiry",
        "stable UUID re-signing",
        "Production deployment",
    ):
        assert required in report
    assert "Real cloud E2E | PASS" not in report