feat(entrypoints): rules.yml loader with shipped framework pack (#27)

Author: rahlk

codeanalyzer/entrypoints/rules.py

@@ -0,0 +1,129 @@
+"""Loading and merging of entrypoint rules (#27).
+
+The shipped ``rules.yml`` covers known frameworks; users extend it with
+``--entrypoint-rules``. User rules merge additively and may ``disable:`` a
+shipped rule by id. A malformed user file is a hard error before analysis
+starts -- silently skipping it would let someone ship rules they believe
+are live.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any, Dict, Iterable, List, Optional
+
+import yaml
+
+_SHIPPED = Path(__file__).with_name("rules.yml")
+_CONFIDENCE = {"declared", "certain", "heuristic"}
+
+
+class RulesError(Exception):
+    """Raised for a malformed rules file. Never swallowed."""
+
+
+@dataclass
+class DecoratorRule:
+    id: str
+    match: str
+    confidence: str = "certain"
+    route: Optional[Dict[str, Any]] = None
+    methods: Optional[Dict[str, Any]] = None
+
+
+@dataclass
+class BaseRule:
+    id: str
+    match: str
+    confidence: str = "certain"
+    transitive: bool = False
+    dispatch: List[str] = field(default_factory=list)
+
+
+@dataclass
+class Framework:
+    name: str
+    detect: List[str] = field(default_factory=list)
+    decorators: List[DecoratorRule] = field(default_factory=list)
+    bases: List[BaseRule] = field(default_factory=list)
+
+
+@dataclass
+class RuleSet:
+    frameworks: Dict[str, Framework] = field(default_factory=dict)
+    rulesets: List[str] = field(default_factory=list)
+
+
+def load_rules(user_paths: Iterable[Path] = ()) -> RuleSet:
+    out = RuleSet()
+    _merge(out, _read(_SHIPPED), "shipped")
+    for p in user_paths:
+        _merge(out, _read(Path(p)), f"user:{p}")
+    return out
+
+
+def _read(path: Path) -> Dict[str, Any]:
+    try:
+        data = yaml.safe_load(path.read_text())
+    except FileNotFoundError as exc:
+        raise RulesError(f"rules file not found: {path}") from exc
+    except yaml.YAMLError as exc:
+        raise RulesError(f"{path}: invalid YAML: {exc}") from exc
+    if not isinstance(data, dict):
+        raise RulesError(f"{path}: top level must be a mapping")
+    return data
+
+
+def _merge(out: RuleSet, data: Dict[str, Any], origin: str) -> None:
+    out.rulesets.append(origin)
+    disabled = set(data.get("disable") or [])
+    frameworks = data.get("frameworks") or {}
+    if not isinstance(frameworks, dict):
+        raise RulesError(f"{origin}: `frameworks` must be a mapping")
+
+    for name, body in frameworks.items():
+        if not isinstance(body, dict):
+            raise RulesError(f"{origin}: framework `{name}` must be a mapping")
+        fw = out.frameworks.setdefault(name, Framework(name=name))
+        fw.detect = sorted(set(fw.detect) | set(body.get("detect") or []))
+        for raw in body.get("decorators") or []:
+            fw.decorators.append(_decorator_rule(raw, origin))
+        for raw in body.get("bases") or []:
+            fw.bases.append(_base_rule(raw, origin))
+
+    for fw in out.frameworks.values():
+        fw.decorators = [r for r in fw.decorators if r.id not in disabled]
+        fw.bases = [r for r in fw.bases if r.id not in disabled]
+
+
+def _require(raw: Dict[str, Any], key: str, origin: str) -> Any:
+    if key not in raw:
+        raise RulesError(f"{origin}: rule {raw!r} is missing `{key}`")
+    return raw[key]
+
+
+def _confidence(raw: Dict[str, Any], origin: str) -> str:
+    c = raw.get("confidence", "certain")
+    if c not in _CONFIDENCE:
+        raise RulesError(f"{origin}: confidence must be one of {sorted(_CONFIDENCE)}, got {c!r}")
+    return c
+
+
+def _decorator_rule(raw: Dict[str, Any], origin: str) -> DecoratorRule:
+    return DecoratorRule(
+        id=_require(raw, "id", origin),
+        match=_require(raw, "match", origin),
+        confidence=_confidence(raw, origin),
+        route=raw.get("route"),
+        methods=raw.get("methods"),
+    )
+
+
+def _base_rule(raw: Dict[str, Any], origin: str) -> BaseRule:
+    return BaseRule(
+        id=_require(raw, "id", origin),
+        match=_require(raw, "match", origin),
+        confidence=_confidence(raw, origin),
+        transitive=bool(raw.get("transitive", False)),
+        dispatch=list(raw.get("dispatch") or []),
+    )

codeanalyzer/entrypoints/rules.yml

@@ -0,0 +1,67 @@
+version: 1
+
+frameworks:
+  flask:
+    detect: [flask]
+    decorators:
+      - id: flask.route
+        match: "flask.Flask.route"
+        route: {from: positional, index: 0}
+        methods: {from: keyword, name: methods, default: [GET]}
+      - id: flask.bp-verb
+        match: "flask.Blueprint.{get,post,put,delete,patch}"
+        route: {from: positional, index: 0}
+        methods: {from: match_suffix}
+    bases:
+      - id: flask.methodview
+        match: "flask.views.MethodView"
+        transitive: true
+        dispatch: [get, post, put, delete, patch]
+
+  fastapi:
+    detect: [fastapi]
+    decorators:
+      - id: fastapi.verb
+        match: "fastapi.FastAPI.{get,post,put,delete,patch,head,options}"
+        route: {from: positional, index: 0}
+        methods: {from: match_suffix}
+      - id: fastapi.router-verb
+        match: "fastapi.APIRouter.{get,post,put,delete,patch}"
+        route: {from: positional, index: 0}
+        methods: {from: match_suffix}
+      - id: fastapi.websocket
+        match: "fastapi.FastAPI.websocket"
+        route: {from: positional, index: 0}
+
+  celery:
+    detect: [celery]
+    decorators:
+      - id: celery.shared-task
+        match: "celery.shared_task"
+      - id: celery.task
+        match: "celery.Celery.task"
+
+  click:
+    detect: [click, typer]
+    decorators:
+      - id: click.command
+        match: "click.{command,group}"
+      - id: typer.command
+        match: "typer.Typer.command"
+
+  drf:
+    detect: [rest_framework]
+    decorators:
+      - id: drf.api-view
+        match: "rest_framework.decorators.api_view"
+      - id: drf.action
+        match: "rest_framework.decorators.action"
+    bases:
+      - id: drf.apiview
+        match: "rest_framework.views.APIView"
+        transitive: true
+        dispatch: [get, post, put, patch, delete, head, options]
+      - id: drf.viewset
+        match: "rest_framework.viewsets.*"
+        transitive: true
+        dispatch: [list, retrieve, create, update, partial_update, destroy]

pyproject.toml

@@ -47,6 +47,9 @@ dependencies = [
     # (scalpel/SSA/const.py uses astor.to_source). Pure-Python, installs
     # everywhere; scalpel pins ~=0.8.1.
     "astor>=0.8.1,<0.9.0",
+    # pyyaml: the entrypoint rules pack (#27) is YAML. Declared explicitly --
+    # it was previously only reachable as a transitive dep of ray.
+    "pyyaml>=6.0,<7.0",
 ]
 
 [project.optional-dependencies]
@@ -56,6 +59,9 @@ neo4j = [
   "neo4j>=5.0.0,<6.0.0",
 ]
 
+[tool.setuptools.package-data]
+codeanalyzer = ["entrypoints/*.yml"]
+
 [dependency-groups]
 test = [
   "pytest>=7.0.0,<8.0.0",

test/test_entrypoint_rules.py

@@ -0,0 +1,26 @@
+import pytest
+
+from codeanalyzer.entrypoints.rules import RulesError, load_rules
+
+
+def test_shipped_rules_load_and_include_flask():
+    rs = load_rules()
+    assert "flask" in rs.frameworks
+    flask = rs.frameworks["flask"]
+    assert "flask" in flask.detect
+    assert any(r.id == "flask.route" for r in flask.decorators)
+
+
+def test_every_shipped_rule_has_a_stable_id_and_valid_confidence():
+    rs = load_rules()
+    for fw in rs.frameworks.values():
+        for rule in list(fw.decorators) + list(fw.bases):
+            assert rule.id, "every rule needs a stable id so users can disable it"
+            assert rule.confidence in {"declared", "certain", "heuristic"}
+
+
+def test_malformed_user_file_raises_before_analysis(tmp_path):
+    bad = tmp_path / "bad.yml"
+    bad.write_text("frameworks: [this is a list not a mapping]\n")
+    with pytest.raises(RulesError):
+        load_rules([bad])