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"] == 8
    dependencies = checks["ID.RA-dependency-vulnerability-management"]["evidence"]
    assert dependencies["sbom_files"] == []
    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_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"