ewoooc/tests/test_security_governance_review_service.py

import json
import subprocess
import sys
from pathlib import Path

import pytest
from flask import Blueprint, Flask, jsonify


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


def test_security_governance_review_is_honest_and_release_gate_passes():
    from services.security_governance_review_service import (
        build_security_governance_review,
    )

    report = build_security_governance_review(root=ROOT, detail_limit=12)

    assert report["status"] == "partial"
    assert report["release_gate"]["status"] == "pass"
    assert report["release_gate"]["blocking_failure_count"] == 0
    assert report["completion"]["asset_field_coverage_percent"] == 100.0
    assert report["completion"]["asset_runtime_reconciliation_complete"] is False
    assert report["completion"]["runtime_closure_passed"] < report["completion"]["runtime_closure_total"]
    assert report["work_items"][0]["id"] == "SEC-P0-001"
    assert len(report["work_items"]) == 22
    assert report["work_items"][9]["id"] == "REL-P0-001"
    assert any(item["id"] == "QA-P1-001" for item in report["work_items"])
    assert report["work_items"][-1]["id"] == "CHAOS-P2-001"
    checks = {item["id"]: item for item in report["checks"]}
    access = checks["PR.AA-route-access-control"]["evidence"]
    assert access["mutating_unguarded_count"] == 0
    assert access["unclassified_unguarded_count"] == 0
    assert access["unguarded_count"] == 7
    dependencies = checks["ID.RA-dependency-vulnerability-management"]["evidence"]
    assert dependencies["sbom_files"] == []
    supply = checks["GV.SC-gitea-only-supply-chain"]
    assert supply["status"] == "pass"
    assert supply["evidence"]["docker_build_context"]["safe"] is True
    quality = checks["GV.QA-test-governance"]
    assert quality["status"] == "partial"
    assert quality["evidence"]["collection_error_count"] == 1
    assert quality["evidence"]["full_suite_passed"] == 1701
    assert report["safety"] == {
        "read_only": True,
        "network_calls": False,
        "database_reads": False,
        "database_writes": False,
        "secret_reads": False,
        "external_side_effects": False,
    }


def test_security_governance_cli_strict_mode_passes_source_gate():
    completed = subprocess.run(
        [
            sys.executable,
            str(ROOT / "scripts" / "ops" / "report_security_governance_review.py"),
            "--root",
            str(ROOT),
            "--compact",
            "--strict",
        ],
        cwd=ROOT,
        capture_output=True,
        text=True,
        timeout=30,
        check=False,
    )

    assert completed.returncode == 0, completed.stderr or completed.stdout
    payload = json.loads(completed.stdout)
    assert payload["release_gate"]["status"] == "pass"
    assert payload["status"] == "partial"


def test_security_governance_cli_can_emit_commit_bound_source_receipt():
    source_commit = "a" * 40
    completed = subprocess.run(
        [
            sys.executable,
            str(ROOT / "scripts" / "ops" / "report_security_governance_review.py"),
            "--root",
            str(ROOT),
            "--strict",
            "--source-commit",
            source_commit,
        ],
        cwd=ROOT,
        capture_output=True,
        text=True,
        timeout=30,
        check=False,
    )

    assert completed.returncode == 0, completed.stderr or completed.stdout
    payload = json.loads(completed.stdout)
    assert payload["receipt_kind"] == "security_governance_source_receipt"
    assert payload["source_commit"] == source_commit
    assert payload["release_gate"]["status"] == "pass"


def test_runtime_subset_uses_commit_bound_source_receipt(tmp_path):
    from services.security_governance_review_service import (
        build_security_governance_review,
    )

    source_report = build_security_governance_review(root=ROOT)
    source_report["receipt_kind"] = "security_governance_source_receipt"
    source_report["source_commit"] = "b" * 40
    governance = tmp_path / "governance"
    governance.mkdir()
    (governance / "security_governance_source_receipt.json").write_text(
        json.dumps(source_report, ensure_ascii=False),
        encoding="utf-8",
    )

    runtime_report = build_security_governance_review(root=tmp_path)
    checks = {item["id"]: item for item in runtime_report["checks"]}

    supply = checks["GV.SC-gitea-only-supply-chain"]
    policy = checks["GV.PO-policy-alignment"]
    assert supply["status"] == "pass"
    assert supply["evidence"]["evidence_source"] == "cd_source_receipt"
    assert policy["status"] == "pass"
    assert policy["evidence"]["evidence_source"] == "cd_source_receipt"


def test_runtime_receipt_advances_access_readback_without_hiding_remaining_debt(tmp_path):
    from services.security_governance_review_service import (
        _runtime_receipt,
    )

    governance = tmp_path / "governance"
    governance.mkdir()
    payload = {
        "receipt_kind": "security_governance_runtime_receipt",
        "trace_id": "secgov-test",
        "run_id": "secgov-test-run",
        "work_item_id": "SEC-P0-001",
        "source_commit": "c" * 40,
        "deployed_version": "V10.773",
        "deployment_mode": "controlled_manual_fallback",
        "access_control": {
            "status": "pass",
            "sensitive_routes_denied": True,
            "security_headers_passed": True,
            "secure_cookie": True,
            "sensitive_route_count": 8,
            "denied_route_count": 8,
            "metrics_public": False,
            "metrics_internal_canary": True,
            "metrics_target_health": "up",
        },
        "database": {"unchanged": True},
        "post_verifier": {
            "health_status": "healthy",
            "governance_gate": "pass",
        },
        "cd": {
            "run_id": 1095,
            "status": "waiting",
            "blocker": "no_ewoooc_host_runner",
        },
        "learning_recorded": False,
        "next_machine_action": "inventory_active_users_then_shadow_database_auth_before_cutover",
    }
    (governance / "security_governance_runtime_receipt.json").write_text(
        json.dumps(payload, ensure_ascii=False),
        encoding="utf-8",
    )

    receipt = _runtime_receipt(tmp_path)

    assert receipt["security_readback_passed"] is True
    assert receipt["database_unchanged"] is True
    assert receipt["metrics_public"] is False
    assert receipt["metrics_internal_canary"] is True
    assert receipt["metrics_target_health"] == "up"
    assert receipt["cd_status"] == "waiting"
    assert receipt["learning_recorded"] is False


def test_runtime_receipt_rejects_public_metrics_even_when_other_checks_pass(tmp_path):
    from services.security_governance_review_service import _runtime_receipt

    governance = tmp_path / "governance"
    governance.mkdir()
    payload = {
        "receipt_kind": "security_governance_runtime_receipt",
        "source_commit": "d" * 40,
        "access_control": {
            "status": "pass",
            "sensitive_routes_denied": True,
            "security_headers_passed": True,
            "secure_cookie": True,
            "metrics_public": True,
            "metrics_internal_canary": True,
        },
        "database": {"unchanged": True},
        "post_verifier": {"health_status": "healthy", "governance_gate": "pass"},
    }
    (governance / "security_governance_runtime_receipt.json").write_text(
        json.dumps(payload),
        encoding="utf-8",
    )

    receipt = _runtime_receipt(tmp_path)

    assert receipt["security_readback_passed"] is False


def test_sensitive_blueprint_policy_denies_anonymous_api(monkeypatch):
    import auth
    from services.http_access_policy_service import enforce_http_access_policy

    monkeypatch.setattr(auth, "DISABLE_LOGIN", False)
    app = Flask(__name__)
    app.secret_key = "test"
    vendor = Blueprint("vendor", __name__)

    @vendor.route("/vendor-stockout/api/vendor/list")
    def list_vendors():
        return jsonify({"success": True})

    app.register_blueprint(vendor)
    app.before_request(enforce_http_access_policy)
    client = app.test_client()

    anonymous = client.get("/vendor-stockout/api/vendor/list")
    assert anonymous.status_code == 401
    assert anonymous.get_json()["error"] == "authentication_required"

    with client.session_transaction() as session:
        session["logged_in"] = True
    authenticated = client.get("/vendor-stockout/api/vendor/list")
    assert authenticated.status_code == 200


def test_public_url_validator_blocks_ssrf_targets(monkeypatch):
    from utils import security

    monkeypatch.setattr(
        security.socket,
        "getaddrinfo",
        lambda *_args, **_kwargs: [(2, 1, 6, "", ("93.184.216.34", 443))],
    )
    assert security.validate_public_http_url("https://example.com/path") == "https://example.com/path"

    monkeypatch.setattr(
        security.socket,
        "getaddrinfo",
        lambda *_args, **_kwargs: [(2, 1, 6, "", ("127.0.0.1", 80))],
    )
    with pytest.raises(ValueError, match="loopback"):
        security.validate_public_http_url("http://example.com/")


def test_alert_auto_fix_protects_database_and_denies_unknown_container(monkeypatch):
    from routes import alert_routes

    def fail_if_called(*_args, **_kwargs):
        raise AssertionError("subprocess must not run for denied containers")

    monkeypatch.setattr(alert_routes.subprocess, "run", fail_if_called)

    protected = alert_routes.auto_fix_container("momo-db", "memory")
    assert protected["success"] is False
    assert "受保護" in protected["message"]

    unknown = alert_routes.auto_fix_container("stock-platform", "memory")
    assert unknown["success"] is False
    assert "白名單" in unknown["message"]


def test_security_governance_smoke_family_is_warning_for_nonblocking_debt(monkeypatch):
    from services import ai_automation_smoke_service as smoke
    from services import security_governance_review_service as review_service

    monkeypatch.setattr(
        review_service,
        "build_security_governance_review",
        lambda **_kwargs: {
            "policy": "test-policy",
            "status": "partial",
            "completion": {
                "program_percent": 45.0,
                "asset_field_coverage_percent": 100.0,
                "asset_runtime_reconciliation_complete": False,
                "runtime_closure_passed": 0,
                "runtime_closure_total": 4,
            },
            "release_gate": {"blocking_failure_count": 0},
            "next_machine_action": "close_access_control",
        },
    )

    check = smoke._security_governance_review_check()

    assert check["status"] == "warning"
    assert check["details"]["blocking_failure_count"] == 0
    assert check["details"]["next_machine_action"] == "close_access_control"