refactor(dedup): one authority per concept for #17856 rows 2, 3, 5, 9-12, 13b, 14, 19 — copies retired, each pinned by a guard (#17856)

autobot-backend/a2a/peer_identity.py

@@ -22,6 +22,7 @@
 from typing import Any, Dict
 from urllib.parse import quote
 
+from autobot_shared.auth.jwt_core import decode_jwt_claims_unverified
 from autobot_shared.logging_manager import get_logger
 from autobot_shared.principal import resolve_principal_id
 
@@ -102,14 +103,7 @@ def jwt_subject_for_audit(authorization: str | None) -> str | None:
 def _decode_jwt_sub(token: str) -> str | None:
     """Decode a JWT's ``sub`` claim without verifying its signature. None on any decode error."""
     try:
-        import base64
-        import json as _json
-
-        parts = token.split(".")
-        if len(parts) != 3:
-            return None
-        claims = _json.loads(base64.urlsafe_b64decode(parts[1] + "=="))
-        return claims.get("sub")
+        return decode_jwt_claims_unverified(token).get("sub")
     except Exception as exc:
         logger.debug("JWT sub decode failed: %s", exc)
         return None

autobot-backend/api/system.py

@@ -762,8 +762,6 @@ async def get_cache_stats(admin_check: bool = Depends(check_admin_permission)):
     Issue #744: Requires admin authentication.
     """
     try:
-        from utils.cache_manager import cache_manager
-
         # Get cache statistics
         cache_stats = await cache_manager.get_stats()
 
@@ -871,8 +869,6 @@ async def get_cache_activity(admin_check: bool = Depends(check_admin_permission)
     Issue #744: Requires admin authentication.
     """
     try:
-        from utils.cache_manager import cache_manager
-
         activity_response = {
             "timestamp": datetime.now(tz=timezone.utc).isoformat(),
             "activity": {

autobot-backend/cache/simple_cache_cold_start_17856_test.py

@@ -0,0 +1,59 @@
+# Copyright 2025-2026 mrveiss
+# SPDX-License-Identifier: Apache-2.0
+# AutoBot - AI-Powered Automation Platform
+# Author: mrveiss
+"""Cold-start cache stats stay "enabled" on the advanced manager (#17856).
+
+The retired `utils.cache_manager.CacheManager.get_stats` awaited `_ensure_redis_client()` first, so
+`GET /cache/stats` on a freshly started process reported the real key count. `SimpleCacheManager`
+had a no-op `_ensure_redis_client`, which would have reported `disabled` until something else
+initialised the client.
+"""
+
+import pytest
+
+from utils.advanced_cache_manager import SimpleCacheManager
+
+
+class _FakeRedis:
+    async def info(self, section=None):
+        return {"used_memory_human": "1M"}
+
+    async def scan_iter(self, match=None):  # pragma: no cover - replaced below
+        raise NotImplementedError
+
+
+class _Advanced:
+    """Stand-in whose client only exists after `_ensure_redis_client` runs."""
+
+    def __init__(self):
+        self.redis_client = None
+        self.ensured = 0
+
+    async def _ensure_redis_client(self):
+        self.ensured += 1
+        client = _FakeRedis()
+
+        async def _scan(match=None):
+            for key in ("cache:a", "cache:b"):
+                yield key
+
+        client.scan_iter = _scan
+        self.redis_client = client
+
+
+@pytest.mark.asyncio
+async def test_get_stats_initialises_the_client_before_reading() -> None:
+    mgr = SimpleCacheManager()
+    mgr._cache = _Advanced()
+    stats = await mgr.get_stats()
+    assert stats == {"status": "enabled", "total_keys": 2, "memory_usage": "1M", "default_ttl": mgr.default_ttl}
+    assert mgr._cache.ensured == 1
+
+
+@pytest.mark.asyncio
+async def test_ensure_redis_client_delegates_to_the_advanced_manager() -> None:
+    mgr = SimpleCacheManager()
+    mgr._cache = _Advanced()
+    await mgr._ensure_redis_client()
+    assert mgr._redis_client is mgr._cache.redis_client is not None

autobot-backend/llm_shared/provider_auth.py

@@ -47,6 +47,7 @@
 from abc import ABC, abstractmethod
 from typing import Any
 
+from autobot_shared.auth.jwt_core import decode_jwt_claims_unverified
 from autobot_shared.logging_manager import get_logger
 
 logger = get_logger(__name__)
@@ -556,22 +557,10 @@ def _decode_jwt_claims(id_token: str) -> dict[str, Any]:
     still rests on the TLS-authenticated token endpoint; this is defence-in-depth
     against a swapped-audience / wrong-issuer token, not a trust anchor.
     """
-    import base64
-    import json
-
-    parts = id_token.split(".")
-    if len(parts) != 3:
-        raise ProviderAuthError("id_token is not a well-formed JWT")
-    payload_b64 = parts[1]
-    payload_b64 += "=" * (-len(payload_b64) % 4)  # restore base64url padding
     try:
-        raw = base64.urlsafe_b64decode(payload_b64.encode("ascii"))
-        claims = json.loads(raw.decode("utf-8"))
-    except Exception as exc:
-        raise ProviderAuthError(f"id_token payload is undecodable: {exc}") from exc
-    if not isinstance(claims, dict):
-        raise ProviderAuthError("id_token payload is not a JSON object")
-    return claims
+        return decode_jwt_claims_unverified(id_token)
+    except ValueError as exc:
+        raise ProviderAuthError(f"id_token {exc}") from exc
 
 
 def validate_token_response(resp: dict, *, provider_name: str, client_id: str) -> None:

autobot-backend/services/rs256_revocation.py

@@ -28,16 +28,17 @@
 import time
 
 from auth_middleware import get_auth_middleware
-from autobot_shared.auth.jwt_core import JWTDecodeError, _peek_alg, decode_jwt_no_verify_exp
+from autobot_shared.auth.jwt_core import (
+    RS256_DENYLIST_PREFIX,
+    JWTDecodeError,
+    _peek_alg,
+    decode_jwt_no_verify_exp,
+)
 from autobot_shared.logging_manager import get_logger
 from autobot_shared.redis_client import get_async_redis_client
 
 logger = get_logger(__name__)
 
-#: Shared Redis key prefix — MUST match ``RS256_DENYLIST_PREFIX`` in
-#: ``autobot-slm-backend/services/rs256_denylist.py``.
-RS256_DENYLIST_PREFIX = "auth:rs256:jti:denylist:"
-
 
 async def revoke_authority_token_jti(token: str) -> bool:
     """Add the jti of an RS256 authority token to the cross-service denylist.

autobot-backend/services/secrets_service.py

@@ -96,14 +96,14 @@ def _init_encryption(self, encryption_key: str | None = None) -> None:
             # 1. Environment variable (direct)
             # 2. Config manager (may map from env)
             # 3. Shared canonical key file (#14081 review round 5): minted
-            #    and persisted if it doesn't exist yet via the same helper
-            #    SecretsManager uses, cross-process-locked, so the two
-            #    classes can never disagree about the key and two processes
-            #    racing a genuine first boot can't each mint their own.
-            #    Replaces the previous fallback here, which generated a key
-            #    on a WARNING and never wrote it to disk -- indistinguishable
-            #    from first boot even with a real store present, and
-            #    undecryptable by any other process or the next restart.
+            #    and persisted if missing via the helper SecretsManager uses,
+            #    cross-process-locked so racing first boots can't each mint
+            #    a key. The classes can still DISAGREE on the key: this one
+            #    prefers AUTOBOT_SECRETS_KEY / security.secrets_key (1, 2),
+            #    api/secrets.py reads only the key file. See #10088.
+            #    Replaces the previous fallback, which generated a key on a
+            #    WARNING and never wrote it to disk -- undecryptable by any
+            #    other process or the next restart.
             env_key = config.secrets_key
             if not env_key:
                 env_key = config_manager.get("security.secrets_key", None)

autobot-backend/tasks/chat_retention.py

@@ -14,16 +14,15 @@
 
 import json
 from datetime import datetime, timedelta, timezone
-from typing import Any, Dict
+from typing import Dict
 
 from autobot_shared.async_compat import run_or_schedule
 from autobot_shared.logging_manager import get_logger
 from autobot_shared.redis_client import get_redis_client
 from autobot_shared.ssot_config import config as ssot_config
 from autobot_shared.ssot_constants import REDIS_KEY
 from celery_app import celery_app
-from services.audit.audit import AuditCategory, AuditEvent
-from services.audit.audit import emit as audit_emit
+from tasks.retention_audit import emit_deletion_audit as _emit_deletion_audit
 
 logger = get_logger(__name__)
 
@@ -68,20 +67,6 @@ def _parse_created_at(raw: bytes) -> datetime | None:
         return None
 
 
-def _emit_deletion_audit(resource_id: str, resource_type: str, extra: Dict[str, Any]) -> None:
-    """Fire-and-forget audit event for a retention deletion."""
-    audit_emit(
-        AuditEvent(
-            category=AuditCategory.SECURITY,
-            action="retention_purge",
-            actor_id="system:retention_worker",
-            resource_type=resource_type,
-            resource_id=resource_id,
-            metadata=extra,
-        )
-    )
-
-
 async def _async_cleanup_expired_chats(retention_days: int, dry_run: bool = False) -> Dict[str, int]:
     """Remove chat sessions/conversations older than retention_days.
 

autobot-backend/tasks/file_retention.py

@@ -13,13 +13,12 @@
 
 from datetime import datetime, timedelta, timezone
 from pathlib import Path
-from typing import Any, Dict
+from typing import Dict
 
 from autobot_shared.logging_manager import get_logger
 from autobot_shared.ssot_config import config as ssot_config
 from celery_app import celery_app
-from services.audit.audit import AuditCategory, AuditEvent
-from services.audit.audit import emit as audit_emit
+from tasks.retention_audit import emit_deletion_audit
 
 logger = get_logger(__name__)
 
@@ -28,20 +27,6 @@
 _STORAGE_BASE = Path(ssot_config.misc.chats_directory or "data") / "uploads"
 
 
-def _emit_deletion_audit(file_path: str, extra: Dict[str, Any]) -> None:
-    """Fire-and-forget audit event for a file retention deletion."""
-    audit_emit(
-        AuditEvent(
-            category=AuditCategory.SECURITY,
-            action="retention_purge",
-            actor_id="system:retention_worker",
-            resource_type="file_attachment",
-            resource_id=file_path,
-            metadata=extra,
-        )
-    )
-
-
 def _cleanup_expired_files(retention_days: int, dry_run: bool = False) -> Dict[str, int]:
     """Remove uploaded files older than retention_days.
 
@@ -80,8 +65,9 @@ def _cleanup_expired_files(retention_days: int, dry_run: bool = False) -> Dict[s
                 deleted_bytes += size
             else:
                 file_path.unlink()
-                _emit_deletion_audit(
+                emit_deletion_audit(
                     str(file_path),
+                    "file_attachment",
                     {"age_days": age_days, "size_bytes": size, "retention_days": retention_days},
                 )
                 deleted_files += 1