feat: dual-mode identity verification and disable semantic challenge by default

Rename verify_signature node to verify_identity with support for both
Ed25519 signatures and OAuth bearer tokens. OAuth validation uses a
conditional import (graceful fallback when airlock.oauth module absent).
Disable semantic challenge by default (challenge_fallback_mode=disabled)
so unknown-reputation agents route directly to issue_verdict.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: shivdeep1

airlock/config.py

@@ -73,7 +73,7 @@ class AirlockConfig(BaseSettings):
     expect_replicas: int = Field(default=1, ge=1)
 
     # Challenge fallback mode when LLM is unavailable: "ambiguous" (default) or "rule_based".
-    challenge_fallback_mode: str = "ambiguous"
+    challenge_fallback_mode: str = "disabled"
 
     # -----------------------------------------------------------------------
     # Scoring (generic defaults — production overrides via env vars)

airlock/engine/orchestrator.py

@@ -3,8 +3,8 @@
 Node map (11 nodes):
   resolve                  -> validate_schema
   validate_schema          -> check_revocation
-  check_revocation         -> verify_signature          (or failed)
-  verify_signature         -> validate_vc               (or failed)
+  check_revocation         -> verify_identity           (or failed)
+  verify_identity          -> validate_vc               (or failed)
   validate_vc              -> validate_delegation        (or failed)
   validate_delegation      -> cross_ref_capabilities    (or failed)
   cross_ref_capabilities   -> check_reputation          (or failed)
@@ -23,9 +23,9 @@
 
 from langgraph.graph import END, StateGraph
 
+from airlock.config import get_config
 from airlock.crypto.keys import KeyPair, resolve_public_key
 from airlock.crypto.signing import sign_attestation, verify_model
-from airlock.config import get_config
 from airlock.crypto.vc import extract_capabilities, validate_credential
 from airlock.engine.state import SessionManager
 from airlock.gateway.revocation import RedisRevocationStore, RevocationStore
@@ -84,6 +84,7 @@ class OrchestrationState(TypedDict, total=False):
     _challenge_outcome: str | None
     _tier: int  # TrustTier int value
     _local_only: bool
+    _bearer_token: str | None
 
 
 # ---------------------------------------------------------------------------
@@ -306,6 +307,7 @@ async def _handle_handshake(self, event: HandshakeReceived) -> None:
             "_challenge_outcome": None,
             "_tier": TrustTier.UNKNOWN,
             "_local_only": getattr(request, "privacy_mode", "any") == "local_only",
+            "_bearer_token": getattr(event, "bearer_token", None),
         }
 
         # Run the graph synchronously through all nodes.
@@ -620,34 +622,64 @@ def _node_check_revocation(self, state: OrchestrationState) -> OrchestrationStat
     def _route_after_revocation(self, state: OrchestrationState) -> str:
         if state.get("failed_at") == "check_revocation":
             return "failed"
-        return "verify_signature"
+        return "verify_identity"
+
+    def _node_verify_identity(self, state: OrchestrationState) -> OrchestrationState:
+        """Node 2: verify agent identity via OAuth bearer token or Ed25519 signature.
 
-    def _node_verify_signature(self, state: OrchestrationState) -> OrchestrationState:
-        """Node 2: verify the Ed25519 signature on the HandshakeRequest."""
+        Dual-mode authentication:
+        1. If a bearer token is present, attempt OAuth validation first.
+        2. If OAuth succeeds and token subject matches initiator DID, mark valid.
+        3. Otherwise fall back to Ed25519 signature verification.
+        """
         checks: list[CheckResult] = list(state.get("check_results", []))
         request = state["handshake"]
+        auth_method = "ed25519"
 
-        try:
-            verify_key = resolve_public_key(request.initiator.did)
-            valid = verify_model(request, verify_key)
-        except Exception as exc:
-            valid = False
-            logger.debug("Signature verification error: %s", exc)
+        # --- Try OAuth bearer token first ---
+        oauth_validated = False
+        bearer_token = state.get("_bearer_token")
+        if bearer_token is not None:
+            try:
+                from airlock.oauth.token_validator import validate_access_token
+
+                gateway_vk = (
+                    self._airlock_keypair.verify_key if self._airlock_keypair is not None else None
+                )
+                token_data = validate_access_token(bearer_token, gateway_vk)
+                if token_data.get("sub") == request.initiator.did:
+                    oauth_validated = True
+                    auth_method = "oauth"
+            except ImportError:
+                logger.debug("OAuth module not available, falling back to Ed25519")
+            except Exception as exc:
+                logger.debug("OAuth token validation failed: %s", exc)
+
+        # --- Fall back to Ed25519 signature verification ---
+        valid = oauth_validated
+        if not valid:
+            try:
+                verify_key = resolve_public_key(request.initiator.did)
+                valid = verify_model(request, verify_key)
+            except Exception as exc:
+                valid = False
+                logger.debug("Signature verification error: %s", exc)
 
+        detail = f"{auth_method} identity verified" if valid else "Identity verification failed"
         checks.append(
             CheckResult(
                 check=VerificationCheck.SIGNATURE,
                 passed=valid,
-                detail="Ed25519 signature valid" if valid else "Signature verification failed",
+                detail=detail,
             )
         )
         state["check_results"] = checks
         state["_sig_valid"] = valid
         if valid:
             state["session"].state = VerificationState.SIGNATURE_VERIFIED
         else:
-            state["error"] = "Invalid signature"
-            state["failed_at"] = "verify_signature"
+            state["error"] = "Invalid identity credentials"
+            state["failed_at"] = "verify_identity"
             state["verdict"] = TrustVerdict.REJECTED
             state["session"].state = VerificationState.FAILED
         return state
@@ -1055,7 +1087,7 @@ def _node_failed(self, state: OrchestrationState) -> OrchestrationState:
     # Conditional edge functions
     # ------------------------------------------------------------------
 
-    def _route_after_signature(self, state: OrchestrationState) -> str:
+    def _route_after_identity(self, state: OrchestrationState) -> str:
         return "validate_vc" if state.get("_sig_valid") else "failed"
 
     def _route_after_vc(self, state: OrchestrationState) -> str:
@@ -1068,6 +1100,9 @@ def _route_after_reputation(self, state: OrchestrationState) -> str:
         elif routing in ("fast_path", "issue_verdict"):
             return "issue_verdict"
         else:
+            cfg = get_config()
+            if cfg.challenge_fallback_mode == "disabled":
+                return "issue_verdict"
             return "semantic_challenge"
 
     # ------------------------------------------------------------------
@@ -1079,7 +1114,7 @@ def _build_graph(self) -> Any:
 
         graph.add_node("validate_schema", self._node_validate_schema)
         graph.add_node("check_revocation", self._node_check_revocation)
-        graph.add_node("verify_signature", self._node_verify_signature)
+        graph.add_node("verify_identity", self._node_verify_identity)
         graph.add_node("validate_vc", self._node_validate_vc)
         graph.add_node("validate_delegation", self._node_validate_delegation)
         graph.add_node("cross_ref_capabilities", self._node_cross_ref_capabilities)
@@ -1095,11 +1130,11 @@ def _build_graph(self) -> Any:
         graph.add_conditional_edges(
             "check_revocation",
             self._route_after_revocation,
-            {"verify_signature": "verify_signature", "failed": "failed"},
+            {"verify_identity": "verify_identity", "failed": "failed"},
         )
         graph.add_conditional_edges(
-            "verify_signature",
-            self._route_after_signature,
+            "verify_identity",
+            self._route_after_identity,
             {"validate_vc": "validate_vc", "failed": "failed"},
         )
         graph.add_conditional_edges(

airlock/gateway/handlers.py

@@ -67,6 +67,14 @@ def _is_valid_did(did: str) -> bool:
 logger = logging.getLogger(__name__)
 
 
+def _extract_bearer_token(request: Request) -> str | None:
+    """Extract an OAuth bearer token from the Authorization header, if present."""
+    auth_header = request.headers.get("authorization", "")
+    if auth_header.lower().startswith("bearer "):
+        return auth_header[7:].strip()
+    return None
+
+
 def _audit_bg(request: Request, **kwargs: object) -> None:
     """Fire-and-forget audit trail append (non-blocking)."""
     trail = getattr(request.app.state, "audit_trail", None)
@@ -310,13 +318,15 @@ async def handle_handshake(
     )
 
     # Publish to event bus — orchestrator handles the rest asynchronously
+    bearer_token = _extract_bearer_token(request)
     event_bus = request.app.state.event_bus
     if not event_bus.try_publish(
         HandshakeReceived(
             session_id=session_id,
             timestamp=datetime.now(UTC),
             request=body,
             callback_url=callback_url,
+            bearer_token=bearer_token,
         )
     ):
         return _nack(request, "Event queue saturated", "SERVICE_BUSY", session_id)

airlock/schemas/events.py

@@ -26,6 +26,7 @@ class HandshakeReceived(VerificationEvent):
     event_type: Literal["handshake_received"] = "handshake_received"
     request: HandshakeRequest
     callback_url: str | None = None
+    bearer_token: str | None = None
 
 
 class SignatureVerified(VerificationEvent):

tests/test_a2a_gateway.py

@@ -409,10 +409,20 @@ async def test_verify_challenge_path_has_challenge_payload(
             target_kp.did,
             text="Semantic path",
         )
-        async with AsyncClient(
-            transport=ASGITransport(app=a2a_app), base_url="http://test"
-        ) as client:
-            resp = await client.post("/a2a/verify", json=body)
+        # Re-enable challenge mode for this test (default is now "disabled")
+        from unittest.mock import patch
+
+        with patch("airlock.engine.orchestrator.get_config") as mock_cfg:
+            from airlock.config import get_config
+
+            real_cfg = get_config()
+            mock_cfg.return_value = real_cfg.model_copy(
+                update={"challenge_fallback_mode": "ambiguous"}
+            )
+            async with AsyncClient(
+                transport=ASGITransport(app=a2a_app), base_url="http://test"
+            ) as client:
+                resp = await client.post("/a2a/verify", json=body)
         data = resp.json()
         assert data["verdict"] == "DEFERRED"
         assert data["challenge"] is not None

tests/test_decay_on_read.py

@@ -101,7 +101,12 @@ async def on_challenge(sid: str, _ch: ChallengeRequest) -> None:
         expires_at=now + timedelta(minutes=2),
     )
 
-    with patch(
+    # Re-enable challenge mode (default is now "disabled")
+    from airlock.config import AirlockConfig
+
+    challenge_cfg = AirlockConfig(challenge_fallback_mode="ambiguous")
+
+    with patch("airlock.engine.orchestrator.get_config", return_value=challenge_cfg), patch(
         "airlock.engine.orchestrator.generate_challenge",
         new=AsyncMock(return_value=fake_ch),
     ):