Grant the trace-forward capability from an operator allowlist, not a self-declared role

.github/workflows/ci.yml

@@ -0,0 +1,21 @@
+name: ci
+
+on:
+  push:
+    branches: [main]
+  pull_request:
+
+jobs:
+  tests:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.12"
+      - name: Install
+        run: |
+          python -m pip install --upgrade pip
+          pip install -r requirements.txt pytest
+      - name: Tests
+        run: python -m pytest tests -q

CHANGELOG.md

@@ -1,5 +1,13 @@
 # Changelog
 
+## [Unreleased]
+
+### Changed
+- **Capabilities come from the operator, not the user**: the `trace-forward` capability was written into a token whenever the user's self-declared `role` was `"authority"`. It is now granted only to accounts listed in `HUB_AUTHORITY_ACCOUNTS`; `role` is descriptive and is no longer an authorisation input. (`capabilities.py`, `hub_main.generate_jwt`)
+
+### Added
+- First test suite (`tests/`) and a CI workflow running it. The role test fails against the previous code.
+
 ## [v0.9-review]
 
 ### Added

README.md

@@ -42,6 +42,7 @@ Follow these steps to run the Hub Gateway locally alongside an `ar2` Node and th
 | `SERVER_BASE_URL` | Base URL of the Gateway instance. | `http://127.0.0.1:8000` | No |
 | `REGISTRY_SERVERS` | JSON string defining backend Nodes and their capabilities (country routing). | `[{"url": "http://127.0.0.1:8001", "countries": ["USA", "India"]}]` | No |
 | `WORLD_SHP_FILE` | Path to the `.shp` file used for point-in-polygon country resolution before routing. | *None* | No |
+| `HUB_AUTHORITY_ACCOUNTS` | Comma-separated account emails the operator has accredited as authorities. Only these accounts receive the `trace-forward` capability in their token. A user's self-declared `role` is descriptive and grants nothing. | *None* (no account carries the capability) | No |
 
 ## Endpoints Overview
 

capabilities.py

@@ -0,0 +1,47 @@
+# Licensed under the EUPL, Version 1.2 or – as soon they will be approved by
+# the European Commission - subsequent versions of the EUPL (the "Licence");
+# You may not use this work except in compliance with the Licence.
+# You may obtain a copy of the Licence at:
+# https://joinup.ec.europa.eu/software/page/eupl
+
+"""Which capabilities the hub writes into a token, and who decides.
+
+Until 2026-10 the hub minted the ``trace-forward`` capability whenever a user's
+``role`` was the string ``"authority"``. ``role`` is a free-text field the user
+types at open registration, so the capability was self-assigned. The node still
+demanded a verified grant or authority credential before disclosing anything, so
+nothing leaked on the role alone -- but a permission anyone can give themselves
+is not a permission, and a reader of the token could not tell the difference.
+
+Capabilities now come from an operator-set allowlist and from nothing else.
+``role`` remains a descriptive field and is not an authorisation input.
+"""
+
+import os
+
+TRACE_FORWARD = "trace-forward"
+
+# Comma-separated account emails the operator has accredited as authorities.
+# Unset or empty means no account carries the capability, which is the safe
+# default for a hub that has not been told who its authorities are.
+ALLOWLIST_ENV = "HUB_AUTHORITY_ACCOUNTS"
+
+
+def authority_accounts(env: dict | None = None) -> frozenset[str]:
+    """The accredited account emails, lower-cased and stripped."""
+    source = os.environ if env is None else env
+    raw = source.get(ALLOWLIST_ENV, "") or ""
+    return frozenset(
+        item.strip().lower() for item in raw.split(",") if item.strip()
+    )
+
+
+def capabilities_for(email: str, env: dict | None = None) -> list[str]:
+    """Capabilities to write into a token for ``email``.
+
+    The caller's self-declared role is deliberately not a parameter: there is
+    no way to pass it in, so there is no way to let it back in by accident.
+    """
+    if email and email.strip().lower() in authority_accounts(env):
+        return [TRACE_FORWARD]
+    return []

hub_main.py

@@ -32,6 +32,7 @@
 from user_database import get_user_db, engine, Base
 from user_models import User, SigningKey, SystemKey
 from user_schemas import UserCreate, UserResponse, UserRegistrationResponse, SigningKeyCreate, SigningKeyResponse
+from capabilities import capabilities_for
 
 load_dotenv()
 
@@ -215,8 +216,12 @@ def generate_jwt(email: str, private_key: str, role: str = "farmer") -> dict:
     from datetime import timezone
     expire = datetime.now(timezone.utc) + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
     claims = {"sub": email, "type": "access", "exp": expire}
-    if role == "authority":
-        claims["capabilities"] = ["trace-forward"]
+    # ``role`` is kept in the signature for the two callers below and is not
+    # consulted: capabilities come from the operator allowlist only. See
+    # capabilities.py for why.
+    caps = capabilities_for(email)
+    if caps:
+        claims["capabilities"] = caps
     token = jwt.encode(claims, private_key, algorithm=ALGORITHM, headers={"kid": "hub-key-1"})
     return {"access_token": token, "token_type": "bearer"}
 

tests/conftest.py

@@ -0,0 +1,15 @@
+# Licensed under the EUPL, Version 1.2 or – as soon they will be approved by
+# the European Commission - subsequent versions of the EUPL (the "Licence");
+# You may not use this work except in compliance with the Licence.
+# You may obtain a copy of the Licence at:
+# https://joinup.ec.europa.eu/software/page/eupl
+
+"""hub_main refuses to import without these two variables. The shapefile is
+only opened when a route resolves a country, and the registry list is only
+read when a route proxies, so placeholders are enough for unit tests."""
+
+import os
+
+os.environ.setdefault("WORLD_SHP_FILE", "/nonexistent/world.shp")
+os.environ.setdefault("REGISTRY_SERVERS", '[{"url": "http://127.0.0.1:8001", "countries": ["USA"]}]')
+os.environ.setdefault("DATABASE_URL", "sqlite://")

tests/test_capabilities.py

@@ -0,0 +1,105 @@
+# Licensed under the EUPL, Version 1.2 or – as soon they will be approved by
+# the European Commission - subsequent versions of the EUPL (the "Licence");
+# You may not use this work except in compliance with the Licence.
+# You may obtain a copy of the Licence at:
+# https://joinup.ec.europa.eu/software/page/eupl
+
+"""The trace-forward capability is granted by the operator, not by the user.
+
+Before this change ``generate_jwt`` wrote ``capabilities=["trace-forward"]``
+into the token whenever ``role == "authority"``, and ``role`` was whatever the
+user typed at registration. The first test here fails against that code.
+"""
+
+import os
+import sys
+
+import jwt
+import pytest
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric import rsa
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from capabilities import ALLOWLIST_ENV, TRACE_FORWARD, authority_accounts, capabilities_for  # noqa: E402
+
+
+# --- the pure function ------------------------------------------------------
+
+def test_a_self_declared_role_grants_nothing():
+    # No allowlist at all: nobody is an authority, whatever they called themselves.
+    assert capabilities_for("anyone@example.org", env={}) == []
+
+
+def test_an_allowlisted_account_gets_trace_forward():
+    env = {ALLOWLIST_ENV: "fda.investigator@example.gov"}
+    assert capabilities_for("fda.investigator@example.gov", env=env) == [TRACE_FORWARD]
+
+
+def test_the_allowlist_is_case_and_whitespace_insensitive():
+    env = {ALLOWLIST_ENV: " A@Example.org , b@example.org "}
+    assert authority_accounts(env) == frozenset({"a@example.org", "b@example.org"})
+    assert capabilities_for("a@EXAMPLE.org", env=env) == [TRACE_FORWARD]
+
+
+def test_an_account_not_on_the_list_gets_nothing_even_when_the_list_is_set():
+    env = {ALLOWLIST_ENV: "fda.investigator@example.gov"}
+    assert capabilities_for("farmer@example.org", env=env) == []
+
+
+def test_an_empty_or_blank_allowlist_means_nobody():
+    assert capabilities_for("x@example.org", env={ALLOWLIST_ENV: ""}) == []
+    assert capabilities_for("x@example.org", env={ALLOWLIST_ENV: " , , "}) == []
+
+
+# --- the token the hub actually mints -----------------------------------------
+
+@pytest.fixture(scope="module")
+def rsa_pem():
+    key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
+    priv = key.private_bytes(
+        serialization.Encoding.PEM,
+        serialization.PrivateFormat.TraditionalOpenSSL,
+        serialization.NoEncryption(),
+    ).decode()
+    pub = key.public_key().public_bytes(
+        serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo
+    ).decode()
+    return priv, pub
+
+
+def _claims(token: str, pub: str) -> dict:
+    return jwt.decode(token, pub, algorithms=["RS256"])
+
+
+def test_generate_jwt_ignores_the_role_argument(monkeypatch, rsa_pem):
+    """The control that matters: role="authority" must not produce the capability.
+
+    This is the exact call the login routes make, with the exact value that
+    used to unlock the capability.
+    """
+    monkeypatch.delenv(ALLOWLIST_ENV, raising=False)
+    from hub_main import generate_jwt
+    priv, pub = rsa_pem
+    out = generate_jwt("self.declared@example.org", priv, role="authority")
+    claims = _claims(out["access_token"], pub)
+    assert "capabilities" not in claims, claims
+
+
+def test_generate_jwt_grants_the_capability_from_the_allowlist(monkeypatch, rsa_pem):
+    monkeypatch.setenv(ALLOWLIST_ENV, "fda.investigator@example.gov")
+    from hub_main import generate_jwt
+    priv, pub = rsa_pem
+    out = generate_jwt("fda.investigator@example.gov", priv, role="farmer")
+    claims = _claims(out["access_token"], pub)
+    assert claims["capabilities"] == [TRACE_FORWARD]
+
+
+def test_the_capability_check_can_fire(monkeypatch, rsa_pem):
+    """Guard against the first test passing vacuously: the same decode path must
+    be able to see a capability when one is present."""
+    monkeypatch.setenv(ALLOWLIST_ENV, "seen@example.org")
+    from hub_main import generate_jwt
+    priv, pub = rsa_pem
+    claims = _claims(generate_jwt("seen@example.org", priv)["access_token"], pub)
+    assert claims.get("capabilities") == [TRACE_FORWARD]