tsk-5bup3r [OPEN] Registry tokens never expire: taOS mints JWTs with

tests/test_agent_registry.py

@@ -273,6 +273,8 @@ def test_mint_and_verify(self, tmp_path):
         assert payload["sub"] == "agent-20260609-120000"
         assert payload["iss"] == "taos-registry"
         assert "iat" in payload
+        assert "exp" in payload
+        assert payload["exp"] > payload["iat"]
         assert payload["user_id"] == "user-1"
         assert payload["framework"] == "openclaw"
 
@@ -578,6 +580,42 @@ async def test_revoked_feed_admin_can_read(self, registry_client):
         resp = await registry_client.get("/api/agents/registry/revoked")
         assert resp.status_code == 200
 
+    async def test_renew_token_returns_new_token(self, registry_client):
+        """POST /api/agents/registry/token/renew issues a fresh token."""
+        reg_resp = await registry_client.post(
+            "/api/agents/registry/register",
+            json={"framework": "openclaw", "display_name": "Renew Me"},
+        )
+        assert reg_resp.status_code == 200
+        old_token = reg_resp.json()["token"]
+
+        renew_resp = await registry_client.post(
+            "/api/agents/registry/token/renew",
+            headers={"Authorization": f"Bearer {old_token}"},
+        )
+        assert renew_resp.status_code == 200
+        new_token = renew_resp.json()["token"]
+        assert new_token != old_token
+
+        pubkey_resp = await registry_client.get("/api/agents/registry/pubkey")
+        pub_pem = pubkey_resp.json()["public_key"].encode()
+        old_payload = verify_registry_token(old_token, pub_pem)
+        new_payload = verify_registry_token(new_token, pub_pem)
+        assert new_payload["sub"] == old_payload["sub"]
+        assert new_payload["user_id"] == old_payload["user_id"]
+        assert new_payload["exp"] > new_payload["iat"]
+
+    async def test_renew_token_requires_bearer(self, registry_client):
+        resp = await registry_client.post("/api/agents/registry/token/renew")
+        assert resp.status_code == 401
+
+    async def test_renew_token_rejects_invalid_token(self, registry_client):
+        resp = await registry_client.post(
+            "/api/agents/registry/token/renew",
+            headers={"Authorization": "Bearer not-a-valid-token"},
+        )
+        assert resp.status_code == 401
+
 
 # ---------------------------------------------------------------------------
 # registry_feeds_read scope -- feed token auth

tests/test_agent_registry_store.py

@@ -1,4 +1,5 @@
 import json
+import time
 from datetime import datetime, timezone
 from pathlib import Path
 
@@ -17,6 +18,7 @@
     load_or_create_signing_keypair,
     mint_canonical_id,
     mint_registry_token,
+    renew_registry_token,
     verify_registry_token,
     VALID_STATUSES,
 )
@@ -256,6 +258,14 @@ def test_token_has_iat(self, signing_keypair):
         assert "iat" in payload
         assert isinstance(payload["iat"], int)
 
+    def test_token_has_exp(self, signing_keypair):
+        priv, pub = signing_keypair
+        token = mint_registry_token("agent-008", priv)
+        payload = verify_registry_token(token, pub)
+        assert "exp" in payload
+        assert isinstance(payload["exp"], int)
+        assert payload["exp"] > payload["iat"]
+
 
 # ---------------------------------------------------------------------------
 # Signing keypair persistence
@@ -1256,10 +1266,112 @@ async def test_org_tree_dangling_reports_to_becomes_root(self, store):
 
 @pytest.mark.asyncio
 async def test_get_by_slug_rejects_glob_metacharacters(store):
-    # A caller-supplied member string with GLOB metachars must not reach the
+    # A caller-supplied member string with GLOB metachrs must not reach the
     # matcher: it returns None instead of matching or raising.
     assert await store.get_by_slug("alpha[a-z]") is None
     assert await store.get_by_slug("alpha*") is None
     assert await store.get_by_slug("alpha?") is None
     assert await store.get_by_slug("") is None
     assert await store.get_by_slug("Alpha") is None
+
+
+# ---------------------------------------------------------------------------
+# Token exp claim: minting, verification, renewal, migration window
+# ---------------------------------------------------------------------------
+
+
+def _build_token_no_exp(priv_pem, sub="agent-noexp") -> str:
+    header = _b64url_encode(
+        json.dumps({"alg": "EdDSA", "typ": "JWT"}, separators=(",", ":")).encode()
+    )
+    claims = {
+        "sub": sub,
+        "iss": "taos-registry",
+        "iat": int(time.time()),
+        "jti": "test-jti-noexp",
+    }
+    payload_b64 = _b64url_encode(json.dumps(claims, separators=(",", ":")).encode())
+    signing_input = f"{header}.{payload_b64}".encode()
+    from cryptography.hazmat.primitives.serialization import load_pem_private_key
+    priv = load_pem_private_key(priv_pem, password=None)
+    sig = _b64url_encode(priv.sign(signing_input))
+    return f"{header}.{payload_b64}.{sig}"
+
+
+class TestTokenExpiration:
+    def test_mint_includes_exp(self, signing_keypair):
+        priv, pub = signing_keypair
+        token = mint_registry_token("agent-exp-check", priv)
+        payload = verify_registry_token(token, pub)
+        assert "exp" in payload
+        assert isinstance(payload["exp"], int)
+        assert payload["exp"] > payload["iat"]
+
+    def test_expired_token_is_rejected(self, signing_keypair, monkeypatch):
+        priv, pub = signing_keypair
+        token = mint_registry_token("agent-expired", priv, lifetime_seconds=0)
+        real_now = time.time()
+        monkeypatch.setattr(
+            "tinyagentos.agent_registry_store.time.time",
+            lambda: real_now + 1,
+        )
+        with pytest.raises(ValueError, match="expired"):
+            verify_registry_token(token, pub)
+
+    def test_token_without_exp_accepted_during_migration(self, signing_keypair, monkeypatch):
+        priv, pub = signing_keypair
+        token = _build_token_no_exp(priv)
+        cutoff = time.time() + 3600
+        monkeypatch.setattr(
+            "tinyagentos.agent_registry_store.time.time",
+            lambda: cutoff - 1800,
+        )
+        payload = verify_registry_token(token, pub, allow_no_exp_until=cutoff)
+        assert payload["sub"] == "agent-noexp"
+
+    def test_token_without_exp_rejected_after_migration(self, signing_keypair, monkeypatch):
+        priv, pub = signing_keypair
+        token = _build_token_no_exp(priv)
+        cutoff = time.time() + 3600
+        monkeypatch.setattr(
+            "tinyagentos.agent_registry_store.time.time",
+            lambda: cutoff + 1800,
+        )
+        with pytest.raises(ValueError, match="no exp claim"):
+            verify_registry_token(token, pub, allow_no_exp_until=cutoff)
+
+    def test_decode_fails_closed_without_exp(self, signing_keypair):
+        priv, pub = signing_keypair
+        token = _build_token_no_exp(priv)
+        with pytest.raises(ValueError, match="no exp claim"):
+            verify_registry_token(token, pub)
+
+
+class TestTokenRenewal:
+    def test_renewal_issues_working_token(self, signing_keypair):
+        priv, pub = signing_keypair
+        original = mint_registry_token(
+            "agent-renew", priv, user_id="u1", framework="fw", lifetime_seconds=60
+        )
+        renewed = renew_registry_token(original, pub, priv)
+        payload = verify_registry_token(renewed, pub)
+        assert payload["sub"] == "agent-renew"
+        assert payload["user_id"] == "u1"
+        assert payload["framework"] == "fw"
+        assert payload["exp"] > payload["iat"]
+
+    def test_renewal_preserves_project_id(self, signing_keypair):
+        priv, pub = signing_keypair
+        original = mint_registry_token(
+            "agent-renew-proj", priv, project_id="proj-99", lifetime_seconds=60
+        )
+        renewed = renew_registry_token(original, pub, priv)
+        payload = verify_registry_token(renewed, pub)
+        assert payload["project_id"] == "proj-99"
+
+    def test_renewal_rejects_bad_signature(self, signing_keypair, tmp_path):
+        priv, _ = signing_keypair
+        _, wrong_pub = load_or_create_signing_keypair(tmp_path / "other_keys")
+        token = mint_registry_token("agent-bad-renew", priv)
+        with pytest.raises(ValueError, match="signature verification failed"):
+            renew_registry_token(token, wrong_pub, priv)

tinyagentos/agent_registry_store.py

@@ -263,6 +263,10 @@ def load_or_create_signing_keypair(data_dir: Path) -> tuple[bytes, bytes]:
 # Token minting (compact JWT-style - header.payload.signature, base64url)
 # ---------------------------------------------------------------------------
 
+DEFAULT_REGISTRY_TOKEN_LIFETIME = 86400  # 24 hours: long enough for multi-turn agent
+# tasks, short enough that a leaked bearer credential is not permanent.
+
+
 def _b64url_encode(data: bytes) -> str:
     import base64
     return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii")
@@ -276,13 +280,41 @@ def _b64url_decode(s: str) -> bytes:
     return base64.urlsafe_b64decode(s)
 
 
+def _verify_signature_only(token: str, public_key_pem: bytes) -> dict:
+    """Verify EdDSA signature and return payload without checking exp.
+
+    Internal helper for the renewal path, which must accept expired tokens
+    so an agent can rotate its credential without human intervention.
+    """
+    from cryptography.hazmat.primitives.serialization import load_pem_public_key
+    from cryptography.exceptions import InvalidSignature
+
+    parts = token.split(".")
+    if len(parts) != 3:
+        raise ValueError("token must have three dot-separated parts")
+
+    header_b64, payload_b64, sig_b64 = parts
+    signing_input = f"{header_b64}.{payload_b64}".encode()
+    sig_bytes = _b64url_decode(sig_b64)
+
+    public_key = load_pem_public_key(public_key_pem)
+    try:
+        public_key.verify(sig_bytes, signing_input)
+    except InvalidSignature:
+        raise ValueError("token signature verification failed") from None
+
+    payload = json.loads(_b64url_decode(payload_b64))
+    return payload
+
+
 def mint_registry_token(
     canonical_id: str,
     private_key_pem: bytes,
     *,
     user_id: str = "",
     framework: str = "",
     project_id: Optional[str] = None,
+    lifetime_seconds: int | None = None,
 ) -> str:
     """Return a signed compact EdDSA JWT: <header>.<payload>.<signature> (base64url).
 
@@ -294,6 +326,7 @@ def mint_registry_token(
       sub        - canonical_id (immutable agent identity)
       iss        - "taos-registry"
       iat        - unix timestamp of issuance
+      exp        - unix timestamp of expiry (iat + lifetime_seconds)
       user_id    - owning user_id at registration time
       framework  - agent framework at registration time
       project_id - project binding, present only when non-empty; absent means
@@ -308,10 +341,13 @@ def mint_registry_token(
     header = _b64url_encode(
         json.dumps({"alg": "EdDSA", "typ": "JWT"}, separators=(",", ":")).encode()
     )
+    if lifetime_seconds is None:
+        lifetime_seconds = DEFAULT_REGISTRY_TOKEN_LIFETIME
     claims: dict = {
         "sub": canonical_id,
         "iss": "taos-registry",
         "iat": int(time.time()),
+        "exp": int(time.time()) + lifetime_seconds,
         "jti": uuid.uuid4().hex,
         "user_id": user_id,
         "framework": framework,
@@ -326,31 +362,53 @@ def mint_registry_token(
     return f"{header}.{payload}.{signature}"
 
 
-def verify_registry_token(token: str, public_key_pem: bytes) -> dict:
+def verify_registry_token(
+    token: str,
+    public_key_pem: bytes,
+    allow_no_exp_until: float | None = None,
+) -> dict:
     """Verify *token* against *public_key_pem*.
 
     Returns the decoded payload dict on success.
-    Raises ``ValueError`` on invalid format or bad signature.
+    Raises ``ValueError`` on invalid format, bad signature, missing exp (after
+    the migration window), or expired token.
     """
-    from cryptography.hazmat.primitives.serialization import load_pem_public_key
-    from cryptography.exceptions import InvalidSignature
+    payload = _verify_signature_only(token, public_key_pem)
 
-    parts = token.split(".")
-    if len(parts) != 3:
-        raise ValueError("token must have three dot-separated parts")
+    now = time.time()
+    exp = payload.get("exp")
+    if exp is not None:
+        if now >= exp:
+            raise ValueError("token has expired")
+    elif allow_no_exp_until is None or now >= allow_no_exp_until:
+        raise ValueError("token has no exp claim")
 
-    header_b64, payload_b64, sig_b64 = parts
-    signing_input = f"{header_b64}.{payload_b64}".encode()
-    sig_bytes = _b64url_decode(sig_b64)
+    return payload
 
-    public_key = load_pem_public_key(public_key_pem)
-    try:
-        public_key.verify(sig_bytes, signing_input)
-    except InvalidSignature:
-        raise ValueError("token signature verification failed") from None
 
-    payload = json.loads(_b64url_decode(payload_b64))
-    return payload
+def renew_registry_token(
+    token: str,
+    public_key_pem: bytes,
+    private_key_pem: bytes,
+    *,
+    lifetime_seconds: int | None = None,
+) -> str:
+    """Renew an existing registry token.
+
+    Verifies the EdDSA signature (but does not require the token to be
+    unexpired) and returns a freshly-minted token with the same claims and
+    a new exp.  This is the self-service path for agents whose token has
+    expired without human re-minting.
+    """
+    payload = _verify_signature_only(token, public_key_pem)
+    return mint_registry_token(
+        payload["sub"],
+        private_key_pem,
+        user_id=payload.get("user_id", ""),
+        framework=payload.get("framework", ""),
+        project_id=payload.get("project_id"),
+        lifetime_seconds=lifetime_seconds,
+    )
 
 
 # ---------------------------------------------------------------------------

tinyagentos/agent_token_auth.py

@@ -36,6 +36,13 @@ def _get_store(request: Request):
     return store
 
 
+def _get_migration_cutoff(request: Request) -> float | None:
+    config = getattr(request.app.state, "config", None)
+    if config is None:
+        return None
+    return getattr(config, "registry_token_migration_cutoff_ts", None)
+
+
 def _get_grants_store(request: Request):
     store = getattr(request.app.state, "agent_grants", None)
     if store is None:
@@ -96,7 +103,7 @@ async def _verify_agent_scope(
     # Verify the EdDSA signature using the registry public key.
     _private_pem, public_pem = _get_keypair(request)
     try:
-        payload = verify_registry_token(raw_token, public_pem)
+        payload = verify_registry_token(raw_token, public_pem, allow_no_exp_until=_get_migration_cutoff(request))
     except ValueError:
         raise HTTPException(status_code=401, detail="invalid or malformed registry token")
 
@@ -178,7 +185,7 @@ async def check_agent_identity(request: Request) -> Optional[str]:
 
     _private_pem, public_pem = _get_keypair(request)
     try:
-        payload = verify_registry_token(raw_token, public_pem)
+        payload = verify_registry_token(raw_token, public_pem, allow_no_exp_until=_get_migration_cutoff(request))
     except ValueError:
         raise HTTPException(status_code=401, detail="invalid or malformed registry token")
 

tinyagentos/config.py

@@ -62,6 +62,8 @@ class AppConfig:
     memory_url: str = "http://localhost:7900"
     wallhaven_api_key: str | None = None
     github_app_id: str = ""
+    registry_token_lifetime_seconds: int = 86400
+    registry_token_migration_cutoff_ts: float | None = None
     config_path: Path | None = None
 
     def to_dict(self) -> dict:
@@ -194,6 +196,8 @@ def load_config(path: Path) -> AppConfig:
         github_app_id=str(data.get("github_app_id", "") or ""),
         config_path=path,
         wallhaven_api_key=wallhaven_api_key,
+        registry_token_lifetime_seconds=int(data.get("registry_token_lifetime_seconds", 86400)),
+        registry_token_migration_cutoff_ts=float(data["registry_token_migration_cutoff_ts"]) if "registry_token_migration_cutoff_ts" in data else None,
     )
     if "github_app_private_key" in data:
         global _deprecation_warned_github_key