Open_Ultra_SAST/tests/test_hunt_prover.py

"""Prover integration on instrumented local Git fixtures."""

import subprocess

import pytest

from openultrasast.hunt.producers import Context
from openultrasast.hunt.prover import prove_change


def repository(tmp_path, text, extra=None, lines=None):
    def git(*args):
        return subprocess.run(
            ["git", "-C", str(tmp_path), "-c", "core.hooksPath=/dev/null", *args], check=True, capture_output=True, text=True
        ).stdout.strip()

    git("init", "-q")
    git("config", "user.name", "Fixture")
    git("config", "user.email", "[email protected]")
    (tmp_path / "Fixture.java").write_text("class Plain {}\n")
    git("add", "Fixture.java")
    git("commit", "-qm", "base")
    base = git("rev-parse", "HEAD")
    (tmp_path / "Fixture.java").write_text(text)
    for path, value in (extra or {}).items():
        (tmp_path / path).parent.mkdir(parents=True, exist_ok=True)
        (tmp_path / path).write_text(value)
        git("add", path)
    git("add", "Fixture.java")
    git("commit", "-qm", "head")
    change = dict(
        id="fixture",
        repo="fixture/app",
        kind="ordinary",
        base=base,
        head=git("rev-parse", "HEAD"),
        files=[{"path": "Fixture.java", "head_lines": lines or [[1, len(text.splitlines())]]}],
    )
    return tmp_path, change


@pytest.mark.parametrize("guard,findings", [("", 1), ("Jenkins.get().checkPermission(Jenkins.ADMINISTER);", 0)])
def test_stapler_verdict(tmp_path, guard, findings):
    root, change = repository(
        tmp_path,
        f"""class X extends Descriptor<Thing> {{
 public Object doTestConnection() {{
  {guard}
  return CredentialsProvider.lookupCredentials(Thing.class);
 }}
}}""",
    )
    ctx = Context(root, change)
    result = prove_change(root, change, ctx)
    assert len(result.findings) == findings
    assert result.coverage.operations_checked == 1
    assert result.coverage.operations_discharged == 1 - findings
    assert result.coverage.reached == 1
    assert len(ctx.read("Fixture.java").encode()) > 0
    if findings:
        (finding,) = result.findings
        assert finding["family"] == "access_control" and finding["function"] == "doTestConnection"
        assert finding["id"] == "obligation:protected_read:Fixture.java:4:path_guard"
        assert finding["entry"]["kind"] == "route" and finding["hops"] == [["Fixture.java::doTestConnection", None]]
        assert finding["operation"]["calls"] == ["lookupCredentials"] and finding["missing"] == "path_guard"
        assert finding["evidence"]


def test_unrecognised_is_coverage_only(tmp_path):
    root, change = repository(tmp_path, "class Plain { public Object run() { return lookupCredentials(); } }")
    result = prove_change(root, change)
    assert not result.findings and result.coverage.frameworks_recognised == {}
    assert result.coverage.unchecked_by_reason["framework_unrecognised"] == 1


@pytest.mark.parametrize("guard_position,expected", [("before", 0), ("after", 1), ("none", 1)])
def test_callee_and_caller_guard_order(tmp_path, guard_position, expected):
    guard = "Jenkins.get().checkPermission(Jenkins.ADMINISTER);"
    root, change = repository(
        tmp_path,
        f"""class X extends Descriptor<Thing> {{
 public Object doTestConnection() {{
  {guard if guard_position == "before" else ""}
  Object result = Helper.readSecret();
  {guard if guard_position == "after" else ""}
  return result;
 }}
}}""",
        {"Helper.java": "class Helper { public static Object readSecret() { return lookupCredentials(); } }"},
        [[2, 7]],
    )
    result = prove_change(root, change)
    assert len(result.findings) == expected and result.coverage.operations_checked == 1
    if expected:
        assert result.findings[0]["path"] == "Helper.java"
        assert len(result.findings[0]["hops"]) == 2


def test_touched_helper_walks_to_entry(tmp_path):
    root, change = repository(
        tmp_path,
        """class X extends Descriptor<Thing> {
 public Object doTestConnection() { return readSecret(); }
 public Object readSecret() { return lookupCredentials(); }
}""",
        lines=[[3, 3]],
    )
    result = prove_change(root, change)
    assert len(result.findings) == 1 and len(result.findings[0]["hops"]) == 2


def test_all_paths_are_checked(tmp_path):
    root, change = repository(
        tmp_path,
        """class X extends Descriptor<Thing> {
 public Object doFirst() {
  Jenkins.get().checkPermission(Jenkins.ADMINISTER);
  return readSecret();
 }
 public Object doSecond() { return readSecret(); }
 public Object readSecret() { return lookupCredentials(); }
}""",
        lines=[[7, 7]],
    )
    result = prove_change(root, change)
    assert len(result.findings) == 1 and result.coverage.operations_discharged == 0


def test_generated_output_and_parse_failure_counted(tmp_path):
    root, change = repository(
        tmp_path,
        "class X { public Object broken( { return lookupCredentials(); }",
        {"dist/bundle.java": "class X { public void doTest() {} }"},
    )
    change["files"].append({"path": "dist/bundle.java", "head_lines": [[1, 1]]})
    result = prove_change(root, change)
    assert not result.findings and result.coverage.unchecked_by_reason["parse_failed"] >= 1
    assert result.coverage.unchecked_by_reason["excluded_dir"] == 1


@pytest.mark.parametrize("exits,expected", [(True, 0), (False, 1)])
def test_caller_permission_predicate_requires_exit(tmp_path, exits, expected):
    root, change = repository(
        tmp_path,
        f"""class X extends Descriptor<Thing> {{
 public Object doTestConnection() {{
  boolean allowed = Jenkins.get().hasPermission(Jenkins.ADMINISTER);
  {"if (!allowed) {" if exits else ""}
  {"return null;" if exits else ""}
  {"}" if exits else ""}
  return Helper.readSecret();
 }}
}}""",
        {"Helper.java": "class Helper { public static Object readSecret() { return lookupCredentials(); } }"},
    )
    result = prove_change(root, change)
    assert len(result.findings) == expected and result.coverage.reached == 1
    assert result.coverage.operations_discharged == 1 - expected


def test_declared_public_route_suppresses_premise(tmp_path):
    root, change = repository(
        tmp_path,
        """class X extends Descriptor<Thing> {
 public Object doTestConnection() { return lookupCredentials(); }
}""",
        {".openultrasast/obligations.toml": 'version = 1\n[[route]]\npath = "doTestConnection"\naccess = "public"\n'},
    )
    result = prove_change(root, change)
    assert not result.findings and result.coverage.operations_checked == 1
    assert result.coverage.operations_discharged == 0


def test_nonpublic_unreached_function_in_partly_covered_repo(tmp_path):
    root, change = repository(
        tmp_path,
        "class Plain { public Object run() { return lookupCredentials(); } }",
        {"Exposed.java": "class X extends Descriptor<Thing> { public void doTest() {} }"},
    )
    result = prove_change(root, change)
    assert result.coverage.frameworks_recognised == {"stapler": 1}
    assert result.coverage.unchecked_by_reason["framework_unrecognised"] == 1
    assert not result.findings


@pytest.mark.parametrize("guard,expected", [("", 1), ("checkPermission(ADMINISTER);", 0)])
def test_handler_without_listed_call(tmp_path, guard, expected):
    root, change = repository(
        tmp_path,
        f"""class X extends Descriptor<Thing> {{
 public Object doCheckValue() {{
  {guard}
  return 1;
 }}
}}""",
    )
    result = prove_change(root, change)
    assert len(result.findings) == expected and result.coverage.operations_checked == 1
    if expected:
        assert result.findings[0]["line"] == 2
        assert result.findings[0]["operation"]["id"] == "stapler-handler"


@pytest.mark.parametrize("middleware,expected", [("->middleware(['auth'])", 0), ("->middleware(['throttle:10'])", 1)])
def test_laravel_enforcement_in_prover(tmp_path, middleware, expected):
    root, change = repository(
        tmp_path,
        "class Plain {}",
        {
            "Editor.php": "<?php class Editor extends Controller { public function store() { Product::save(); } }",
            "routes/web.php": f"<?php Route::post('/items', [Editor::class, 'store']){middleware};",
        },
    )
    change["files"] = [{"path": "Editor.php", "head_lines": [[1, 1]]}]
    result = prove_change(root, change)
    assert len(result.findings) == expected and result.coverage.operations_checked == 1
    assert result.coverage.frameworks_recognised == {"laravel": 1}


@pytest.mark.parametrize("arrow", [False, True])
def test_remix_session_does_not_discharge_ownership(tmp_path, arrow):
    declaration = "export const action = async () =>" if arrow else "export async function action()"
    text = f"""{declaration} {{
 const user = await requireUser(request);
 return deactivateRecord(requestId);
}}"""
    root, change = repository(tmp_path, "class Plain {}", {"app/routes/items.ts": text})
    change["files"] = [{"path": "app/routes/items.ts", "head_lines": [[1, 4]]}]
    result = prove_change(root, change)
    assert result.coverage.operations_checked == 1 and result.coverage.operations_discharged == 0
    assert len(result.findings) == 1 and result.findings[0]["missing"] == "ownership_check"


@pytest.mark.parametrize("decorator,expected", [("", 1), ("@tornado.web.authenticated", 0)])
def test_tornado_enforcement_in_prover(tmp_path, decorator, expected):
    text = f"""import tornado.web
class Handler(tornado.web.RequestHandler):
    {decorator}
    def get(self):
        return 1
"""
    root, change = repository(tmp_path, "class Plain {}", {"app.py": text})
    change["files"] = [{"path": "app.py", "head_lines": [[1, 5]]}]
    result = prove_change(root, change)
    assert len(result.findings) == expected and result.coverage.operations_checked == 1


def test_symfony_yaml_is_read_by_prover(tmp_path):
    root, change = repository(
        tmp_path,
        "class Plain {}",
        {
            "Controller.php": (
                "<?php class Editor extends AbstractController { #[Route('/admin')] public function store() { Product::save(); } }"
            ),
            "config/packages/security.yaml": "security:\n  access_control:\n    - { path: ^/admin, roles: ROLE_ADMIN }\n",
        },
    )
    change["files"] = [{"path": "Controller.php", "head_lines": [[1, 1]]}]
    ctx = Context(root, change)
    result = prove_change(root, change, ctx)
    assert not result.findings and result.coverage.operations_discharged == 1
    assert len(ctx.texts["config/packages/security.yaml"].encode()) > 0


@pytest.mark.parametrize("exits,expected", [(True, 0), (False, 1)])
def test_handler_permission_predicate_requires_exit(tmp_path, exits, expected):
    root, change = repository(
        tmp_path,
        f"""class X extends Descriptor<Thing> {{
 public Object doCheckValue() {{
  boolean allowed = hasPermission(ADMINISTER);
  {"if (!allowed) {" if exits else ""}
  {"return null;" if exits else ""}
  {"}" if exits else ""}
  return 1;
 }}
}}""",
    )
    result = prove_change(root, change)
    assert len(result.findings) == expected and result.coverage.operations_discharged == 1 - expected


@pytest.mark.parametrize("public,expected_missing", [(False, ["path_guard", "ownership_check"]), (True, ["ownership_check"])])
def test_remix_missing_kinds_and_public_policy(tmp_path, public, expected_missing):
    extra = {"app/routes/items.ts": "export function action() { return deactivateRecord(requestId); }"}
    if public:
        extra[".openultrasast/obligations.toml"] = 'version = 1\n[[route]]\npath = "action"\naccess = "public"\n'
    root, change = repository(tmp_path, "class Plain {}", extra)
    change["files"] = [{"path": "app/routes/items.ts", "head_lines": [[1, 1]]}]
    result = prove_change(root, change)
    assert [finding["missing"] for finding in result.findings] == expected_missing