feat(authorization-obligations): harvest mode handler_context and the obligation label

Task 4.1. A new excerpt mode keeps the labeled handler with its decorators and
every statement outside it that registers the handler by name, so guards living
in decorators, routers and middleware travel with the pair. Comment and string
mentions never count.

The closed mechanism vocabulary gains the three obligation ids, catalog rows
accept an additive obligation field validated against the closed operation
kinds, and the pair scorer counts a finding tagged with that obligation inside
the labeled function as a detection. Rows without the label run the same code
as before.

Co-Authored-By: Claude Opus 5 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_01T7XD7x5KonMdmGRkp63EQG

tests/test_pair_corpus.py

@@ -985,3 +985,57 @@ def test_vibe_py_and_agent_vfc_carry_pointer_rows_and_the_default_run_skips_them
     payload = json.loads(buf.getvalue())
     skipped = [item for item in payload["degradations"] if item["reason"] == "pointer_pair_skipped"]
     assert len(skipped) == len(pointers) and payload["overall"]["pairs"] == 35
+
+
+_OBLIGATION_HEAD = (
+    '[[pair]]\nname = "leaky-books"\nslice = "vibe-py"\nlanguage = "python"\nvuln = "v.py"\nfixed = "f.py"\nrelpath = "app.py"\n'
+    'provenance = "human"\nreview_tier = "seeded"\n\n'
+    '[[pair.expected]]\ncwe = "CWE-639"\nclass = "broken_access_control"\npath = "app.py"\nfunction = "leaky"\n'
+    'mechanism = "unconstrained_protected_read"\n'
+)
+_LEAKY = (
+    "from flask import request\n\n\n"
+    "@app.route('/books/<title>')\ndef leaky(title):\n"
+    "    return Book.query.filter_by(book_title=title).first()\n"
+)
+_CONSTRAINED = (
+    "from flask import request\n\n\n"
+    "@app.route('/books/<title>')\ndef leaky(title):\n"
+    "    user_id = request.user.id\n"
+    "    return Book.query.filter_by(user_id=user_id, book_title=title).first()\n"
+)
+
+
+def test_obligation_labeled_row_loads_and_is_detected_by_an_obligation_finding(tmp_path: Path) -> None:
+    """authorization-obligations 4.1 (Req 7.2, 7.3): the additive `obligation` label and the scorer's one additive rule."""
+    from openultrasast.pairs import _load_catalog_file, load_mechanisms
+
+    _, ids = load_mechanisms()
+    assert {"unconstrained_protected_read", "unconstrained_protected_write", "unguarded_privileged_action"} <= ids
+    (tmp_path / "v.py").write_text(_LEAKY)
+    (tmp_path / "f.py").write_text(_CONSTRAINED)
+    (tmp_path / "catalog.toml").write_text(_OBLIGATION_HEAD + 'obligation = "protected_read"\n')
+    (case,) = _load_catalog_file(tmp_path / "catalog.toml")
+    assert case.expected[0].obligation == "protected_read"
+    outcome = evaluate_pair(case)
+    assert outcome.detected_vuln and outcome.vuln_matched == 1 and "obligation" in outcome.detection_kinds
+    assert outcome.silent_fix and outcome.pair_correct  # the constrained twin discharges the obligation
+    # the same finding outside the labeled function is not a detection
+    outside = _LEAKY.replace("def leaky(", "def someone_else(")
+    (tmp_path / "v2.py").write_text(outside)
+    (tmp_path / "catalog2.toml").write_text(_OBLIGATION_HEAD.replace('vuln = "v.py"', 'vuln = "v2.py"') + 'obligation = "protected_read"\n')
+    (elsewhere,) = _load_catalog_file(tmp_path / "catalog2.toml")
+    assert not evaluate_pair(elsewhere).detected_vuln
+
+
+def test_catalog_rejects_an_obligation_kind_outside_the_closed_set(tmp_path: Path) -> None:
+    from openultrasast.pairs import CatalogError, _load_catalog_file
+
+    (tmp_path / "v.py").write_text(_LEAKY)
+    (tmp_path / "f.py").write_text(_CONSTRAINED)
+    (tmp_path / "catalog.toml").write_text(_OBLIGATION_HEAD + 'obligation = "guess"\n')
+    with pytest.raises(CatalogError, match="unknown obligation 'guess'"):
+        _load_catalog_file(tmp_path / "catalog.toml")
+    (tmp_path / "plain.toml").write_text(_OBLIGATION_HEAD)  # rows without the field load as before
+    (case,) = _load_catalog_file(tmp_path / "plain.toml")
+    assert case.expected[0].obligation is None

tests/test_pair_harvest.py

@@ -522,3 +522,42 @@ def test_agent_vfc_builder_writes_toml_booleans() -> None:
     b = runpy.run_path(str(Path("benchmarks/pairs/agent-vfc/build_recipes.py")), run_name="agent_builder")
     (row,) = tomllib.loads(b["to_toml"]([{"name": "p", "vendored": False, "line": 3, "reviewer": "pending"}]))["recipe"]
     assert row["vendored"] is False and row["line"] == 3
+
+
+def test_handler_context_mode_keeps_decorators_and_registration() -> None:
+    """authorization-obligations 4.1 (Req 7.1): the handler travels with the statements that register it."""
+    lib = _lib()
+    source = (
+        "from flask import request\n\n"
+        "@app.route('/books/<title>')\n"
+        "@login_required\n"
+        "def guarded(title):\n"
+        "    return Book.query.filter_by(book_title=title).first()\n\n\n"
+        "def other():\n"
+        "    return 1\n\n\n"
+        "app.add_url_rule('/legacy/books/<title>', view_func=guarded)\n"
+        "# guarded is mentioned in a comment and 'guarded' in a string: neither is a registration\n"
+        "label = 'guarded'\n"
+    )
+    excerpt = lib["extract_handler_context"](source, "guarded", "python")
+    assert excerpt.startswith("@app.route('/books/<title>')\n@login_required\ndef guarded(title):")
+    assert "app.add_url_rule('/legacy/books/<title>', view_func=guarded)" in excerpt
+    assert "def other" not in excerpt and "mentioned in a comment" not in excerpt and "label = 'guarded'" not in excerpt
+    js = "function list(req, res) {\n  return Book.find({});\n}\n\nfunction other() {}\n\nrouter.get('/books', requireAuth, list);\n"
+    js_excerpt = lib["extract_handler_context"](js, "list", "javascript")
+    assert "function list(req, res)" in js_excerpt and "router.get('/books', requireAuth, list);" in js_excerpt
+    assert "function other" not in js_excerpt
+    recipe = {
+        "name": "r",
+        "parent": "p",
+        "commit": "c",
+        "path": "app.py",
+        "mode": "handler_context",
+        "function": "guarded",
+        "license": "MIT",
+    }
+    lib["validate_recipe"](recipe)
+    with pytest.raises(lib["RecipeError"], match="function"):
+        lib["validate_recipe"]({**recipe, "function": ""})
+    (vuln, start), (fixed, _) = lib["extract_pair_with_lines"]({**recipe}, source, source.replace("@login_required\n", ""))
+    assert start == 3 and "add_url_rule" in vuln and "@login_required" not in fixed and "add_url_rule" in fixed