fix(annuaire): valide scope (anti path-traversal) à la frontière FS + modèle + DRYRUN ctl

Co-Authored-By: Gerald KERMA <devel@cybermind.fr>

packages/secubox-annuaire/annuaire/config_apply.py

@@ -27,6 +27,7 @@ from __future__ import annotations
 import hashlib
 import json
 import os
+import re
 from pathlib import Path
 from typing import Any, Dict, Iterable, List, Optional
 
@@ -37,6 +38,18 @@ except ImportError:  # pragma: no cover
 
 from .config_compose import compose
 
+# A scope becomes a filesystem path component (<target_dir>/<scope>.toml).
+# It MUST NOT be able to escape target_dir — no '/', no '..', no empty
+# string. This is the filesystem-boundary guard: it is checked BEFORE any
+# Path is ever constructed from an untrusted scope, in both apply_blob
+# (mesh-sourced ConfigBlob) and apply_composed (grant-routed layers).
+_SCOPE_RE = re.compile(r"^[a-z0-9][a-z0-9._-]*$")
+
+
+def _valid_scope(scope: Any) -> bool:
+    """True iff *scope* is safe to use as a bare filename component."""
+    return isinstance(scope, str) and bool(_SCOPE_RE.match(scope))
+
 
 def blob_text(payload: Optional[Dict[str, Any]]) -> Optional[str]:
     """Extract the raw config text from a ConfigBlob payload, or None."""
@@ -83,6 +96,9 @@ def apply_blob(
     if not scope or not isinstance(version, int) or not content_hash:
         return {"status": "reject", "scope": scope, "reason": "malformed-blob"}
 
+    if not _valid_scope(scope):
+        return {"status": "reject", "scope": scope, "reason": "invalid-scope"}
+
     cur = state.get(scope, {})
     if isinstance(cur.get("version"), int) and cur["version"] >= version:
         return {"status": "skip", "scope": scope, "reason": "version-not-newer"}
@@ -131,6 +147,9 @@ def apply_composed(
     previous active → atomic ``os.replace``. Never raises; returns a status
     dict with status in {applied, skip, reject}.
     """
+    if not _valid_scope(scope):
+        return {"status": "reject", "scope": scope, "reason": "invalid-scope"}
+
     try:
         text = compose(list(ordered_layer_texts))
         _toml.loads(text)  # re-validate the composed output before writing

packages/secubox-annuaire/annuaire/model.py

@@ -527,7 +527,12 @@ class ConfigBlob(BaseModel):
 
     config_id:    str = Field(..., description="stable id for this config stream, e.g. 'cfg-<scope>'")
     publisher:    str = Field(..., pattern=r"^did:plc:[0-9a-f]{32}$")
-    scope:        str = Field(..., description="what this configures, e.g. a module name 'yacy'")
+    scope:        str = Field(
+        ...,
+        pattern=r"^[a-z0-9][a-z0-9._-]*$",
+        description="what this configures, e.g. a module name 'yacy' — becomes a bare "
+                    "filename component on disk (config_apply.py), so no '/' or '..'",
+    )
     version:      int = Field(..., ge=0, description="monotonic; higher wins (last-writer-wins)")
     content_hash: str = Field(..., description="BLAKE2b-256 hex of the canonical config content")
     layer:        str = Field(default="baseline", description="config layer; local is box-only")
@@ -563,7 +568,12 @@ class Grant(BaseModel):
     grant_id:   str = Field(..., description="stable id for this grant")
     center_did: str = Field(..., pattern=r"^did:plc:[0-9a-f]{32}$")
     capability: str = Field(default="config", description="what is delegated, e.g. 'config'")
-    scope:      str = Field(..., description="what this grant covers, e.g. a module name 'firewall'")
+    scope:      str = Field(
+        ...,
+        pattern=r"^[a-z0-9][a-z0-9._-]*$",
+        description="what this grant covers, e.g. a module name 'firewall' — becomes a "
+                    "bare filename component on disk (config_apply.py), so no '/' or '..'",
+    )
     layer:      str = Field(..., description="config layer this grant is confined to")
     issued_by:  str = Field(..., pattern=r"^did:plc:[0-9a-f]{32}$")
     created_at: str = Field(default_factory=now_rfc3339)

packages/secubox-annuaire/sbin/sbx-centersctl

@@ -39,6 +39,13 @@ the box's sovereign identity, provisioned once elsewhere):
   CONFIG_TARGET_DIR  default /etc/secubox               (route)
   CONFIG_LOCAL_DIR   default /etc/secubox/config-local   (route)
   ANNUAIRE_LIB       default /usr/lib/secubox/annuaire  (import root, prod)
+
+  DRYRUN=1           on grant/revoke/route: do NOT append to the journal and
+                      do NOT apply any config -- print the action that WOULD
+                      have been taken as {"dryrun": true, "would": ..., ...}
+                      and exit 0. Nothing is written in this mode (grant/
+                      revoke never touch the journal; route never touches
+                      CONFIG_TARGET_DIR); `list` afterwards is unchanged.
 """
 from __future__ import annotations
 
@@ -62,6 +69,22 @@ def _die(msg: str, code: int = 1) -> "None":
     raise SystemExit(code)
 
 
+def _is_dryrun() -> bool:
+    """True iff DRYRUN=1 — grant/revoke/route must not write anything."""
+    return os.environ.get("DRYRUN") == "1"
+
+
+def _dryrun_report(would: str, **fields) -> int:
+    """Print the {"dryrun": true, "would": ...} preview and return rc0.
+
+    Used by grant/revoke/route in DRYRUN mode INSTEAD OF the real write path
+    — no journal append, no apply, nothing touched on disk beyond what a
+    read-only preview needs (e.g. listing the current grant matrix).
+    """
+    print(json.dumps({"dryrun": True, "would": would, **fields}, indent=2))
+    return 0
+
+
 def _load_box_priv() -> bytes:
     """Load the box's own 32-byte raw Ed25519 private key (hex-encoded on disk).
 
@@ -109,6 +132,16 @@ def _box_identity():
 def cmd_grant(args) -> int:
     from annuaire.verbs import grant_issue  # noqa: PLC0415
 
+    if _is_dryrun():
+        from annuaire import grants  # noqa: PLC0415
+        j = _journal()
+        reason = grants.validate_issue(list(j.iter_entries()), args.scope, args.layer)
+        return _dryrun_report(
+            "grant",
+            center_did=args.center_did, scope=args.scope, layer=args.layer,
+            valid=reason is None, reason=reason,
+        )
+
     box_priv, box_did = _box_identity()
     j = _journal()
     try:
@@ -123,6 +156,9 @@ def cmd_grant(args) -> int:
 def cmd_revoke(args) -> int:
     from annuaire.verbs import grant_revoke  # noqa: PLC0415
 
+    if _is_dryrun():
+        return _dryrun_report("revoke", grant_id=args.grant_id)
+
     box_priv, box_did = _box_identity()
     j = _journal()
     try:
@@ -150,6 +186,19 @@ def cmd_list(args) -> int:
 def cmd_route(args) -> int:
     from annuaire.config_router import route_config  # noqa: PLC0415
 
+    if _is_dryrun():
+        from pathlib import Path  # noqa: PLC0415
+
+        from annuaire import grants  # noqa: PLC0415
+
+        j = _journal()
+        grant_map = grants.active_grants(list(j.iter_entries()))
+        scopes = {scope for (scope, _layer) in grant_map.keys()}
+        local_root = Path(CONFIG_LOCAL_DIR)
+        if local_root.is_dir():
+            scopes.update(f.stem for f in local_root.glob("*.toml"))
+        return _dryrun_report("route", scopes=sorted(scopes), target_dir=CONFIG_TARGET_DIR)
+
     # self_did is best-effort (interface symmetry with apply_pending's
     # self-publish loop guard) — route_config works fine without it, and the
     # box key may legitimately be absent on a node that only ever routes

packages/secubox-annuaire/tests/test_apply_composed.py

@@ -23,3 +23,34 @@ def test_apply_composed_bad_toml_keeps_lastgood(tmp_path):
     r = apply_composed("s", ['this is = = not toml\n'], str(tmp_path))
     assert r["status"] == "reject"
     assert (tmp_path/"s.toml").read_text() == before   # last-good untouched
+
+
+# ---------------------------------------------------------------------------
+# path traversal via scope — CRITICAL: scope becomes a filename component,
+# so a scope that walks out of target_dir must be rejected BEFORE any Path
+# is ever built from it.
+# ---------------------------------------------------------------------------
+
+def test_apply_composed_rejects_dotdot_scope(tmp_path):
+    r = apply_composed("../../../../tmp/pwned", ["x=1\n"], str(tmp_path))
+    assert r["status"] == "reject"
+    assert r["reason"] == "invalid-scope"
+    # nothing escaped tmp_path: no .toml anywhere outside it, and tmp_path
+    # itself only ever gets what apply_composed legitimately wrote in other
+    # tests in this module (none here) — assert it stayed empty.
+    assert list(tmp_path.rglob("*.toml")) == []
+    assert not Path("/tmp/pwned.toml").exists()
+
+
+def test_apply_composed_rejects_slash_scope(tmp_path):
+    r = apply_composed("etc/passwd", ["x=1\n"], str(tmp_path))
+    assert r["status"] == "reject"
+    assert r["reason"] == "invalid-scope"
+    assert list(tmp_path.rglob("*.toml")) == []
+
+
+def test_apply_composed_rejects_absolute_path_scope(tmp_path):
+    r = apply_composed("/etc/pwned", ["x=1\n"], str(tmp_path))
+    assert r["status"] == "reject"
+    assert r["reason"] == "invalid-scope"
+    assert not Path("/etc/pwned.toml").exists()

packages/secubox-annuaire/tests/test_centersctl.py

@@ -170,3 +170,74 @@ def test_missing_box_key_fails_clearly(env, center_did):
     err = json.loads(proc.stderr)
     assert "error" in err
     assert "box key" in err["error"]
+
+
+# ---------------------------------------------------------------------------
+# DRYRUN=1 — grant/revoke/route must preview only, writing nothing
+# ---------------------------------------------------------------------------
+
+
+def test_dryrun_grant_writes_nothing(env, center_did):
+    env = {**env, "DRYRUN": "1"}
+    proc = _run(["grant", center_did, "firewall", "baseline"], env)
+    assert proc.returncode == 0, proc.stderr
+    out = json.loads(proc.stdout)
+    assert out["dryrun"] is True
+    assert out["would"] == "grant"
+    assert out["scope"] == "firewall"
+    assert out["layer"] == "baseline"
+    assert out["valid"] is True
+
+    proc = _run(["list"], env)
+    assert proc.returncode == 0, proc.stderr
+    assert json.loads(proc.stdout)["grants"] == []
+
+
+def test_dryrun_grant_reports_invalid_without_dying(env, center_did):
+    """Even a request that WOULD be rejected previews cleanly (rc0, no journal write)."""
+    env = {**env, "DRYRUN": "1"}
+    proc = _run(["grant", center_did, "auth", "baseline"], env)
+    assert proc.returncode == 0, proc.stderr
+    out = json.loads(proc.stdout)
+    assert out["dryrun"] is True
+    assert out["valid"] is False
+    assert out["reason"] == "scope-not-delegatable"
+
+    proc = _run(["list"], env)
+    assert json.loads(proc.stdout)["grants"] == []
+
+
+def test_dryrun_revoke_writes_nothing(env, center_did):
+    proc = _run(["grant", center_did, "firewall", "baseline"], env)
+    assert proc.returncode == 0, proc.stderr
+    grant_id = json.loads(proc.stdout)["grant_id"]
+
+    dry_env = {**env, "DRYRUN": "1"}
+    proc = _run(["revoke", grant_id], dry_env)
+    assert proc.returncode == 0, proc.stderr
+    out = json.loads(proc.stdout)
+    assert out["dryrun"] is True
+    assert out["would"] == "revoke"
+    assert out["grant_id"] == grant_id
+
+    # the real grant is still active — DRYRUN never touched the journal
+    proc = _run(["list"], env)
+    matrix = json.loads(proc.stdout)["grants"]
+    assert len(matrix) == 1
+    assert matrix[0]["scope"] == "firewall"
+
+
+def test_dryrun_route_writes_nothing(env, center_did):
+    proc = _run(["grant", center_did, "firewall", "baseline"], env)
+    assert proc.returncode == 0, proc.stderr
+
+    dry_env = {**env, "DRYRUN": "1"}
+    proc = _run(["route"], dry_env)
+    assert proc.returncode == 0, proc.stderr
+    out = json.loads(proc.stdout)
+    assert out["dryrun"] is True
+    assert out["would"] == "route"
+    assert "firewall" in out["scopes"]
+
+    # nothing was applied to CONFIG_TARGET_DIR
+    assert not Path(env["CONFIG_TARGET_DIR"]).exists()

packages/secubox-annuaire/tests/test_config_apply.py

@@ -82,6 +82,27 @@ def test_apply_rejects_blob_without_inline_text(tmp_path):
     assert r["status"] == "reject" and r["reason"] == "no-inline-text"
 
 
+# ---------------------------------------------------------------------------
+# path traversal via scope — CRITICAL: a mesh-sourced ConfigBlob's scope must
+# not be able to write outside target_dir.
+# ---------------------------------------------------------------------------
+
+def test_apply_blob_rejects_dotdot_scope(tmp_path):
+    state = {}
+    text = "x = 1\n"
+    r = apply_blob(_blob("../../../../tmp/pwned", 1, text), str(tmp_path), state)
+    assert r["status"] == "reject" and r["reason"] == "invalid-scope"
+    assert state == {}
+    assert list(tmp_path.rglob("*.toml")) == []
+
+
+def test_apply_blob_rejects_slash_scope(tmp_path):
+    state = {}
+    r = apply_blob(_blob("etc/passwd", 1, "x = 1\n"), str(tmp_path), state)
+    assert r["status"] == "reject" and r["reason"] == "invalid-scope"
+    assert state == {}
+
+
 # ---------------------------------------------------------------------------
 # apply_pending — allowlist + single-writer gating + state persistence
 # ---------------------------------------------------------------------------

packages/secubox-annuaire/tests/test_grant_model.py

@@ -30,3 +30,33 @@ def test_configblob_has_layer_default_baseline():
     b = ConfigBlob(config_id="cfg-firewall", publisher="did:plc:"+("a"*32),
                    scope="firewall", version=1, content_hash="deadbeef")
     assert b.layer == "baseline"
+
+
+# ---------------------------------------------------------------------------
+# path traversal via scope — CRITICAL, defense in depth: a Grant/ConfigBlob
+# carrying a traversal scope must be invalid AT CONSTRUCTION, before it ever
+# reaches config_apply.py's filesystem-boundary guard.
+# ---------------------------------------------------------------------------
+
+def test_grant_rejects_traversal_scope():
+    import pydantic
+    import pytest
+    with pytest.raises(pydantic.ValidationError):
+        Grant(grant_id="g1", center_did="did:plc:"+("a"*32), capability="config",
+              scope="../evil", layer="baseline", issued_by="did:plc:"+("b"*32))
+
+
+def test_grant_rejects_slash_scope():
+    import pydantic
+    import pytest
+    with pytest.raises(pydantic.ValidationError):
+        Grant(grant_id="g1", center_did="did:plc:"+("a"*32), capability="config",
+              scope="etc/passwd", layer="baseline", issued_by="did:plc:"+("b"*32))
+
+
+def test_configblob_rejects_traversal_scope():
+    import pydantic
+    import pytest
+    with pytest.raises(pydantic.ValidationError):
+        ConfigBlob(config_id="cfg-x", publisher="did:plc:"+("a"*32),
+                   scope="../../etc/pwned", version=1, content_hash="deadbeef")