Centralize capability auth completion normalization

Move callback/code parsing and validation into capability manager, add explicit auth completion error taxonomy, and pass normalized auth payloads to providers/bridge.

Co-Authored-By: GPT-5 Codex <codex@openai.com>

Author: dcramer

specs/capabilities.md

@@ -354,10 +354,18 @@ Request params:
 {
   "flow_id": "caf_01...",
   "callback_url": "https://localhost/callback?code=...",
+  "code": "optional-direct-auth-code",
   "context_token": "<signed-token>"
 }
 ```
 
+Normalization rules (host-owned, provider-independent):
+
+- Accept `code`, `callback_url`, or both.
+- If both are provided and disagree, fail with `capability_auth_code_conflict`.
+- If callback `state` is present and mismatches stored flow state, fail with `capability_auth_state_mismatch`.
+- If no usable code can be resolved, fail with `capability_auth_code_missing`.
+
 Response:
 
 ```json
@@ -426,6 +434,10 @@ See `specs/browser.md` for runtime bridge invariants.
 | Auth flow expired/invalid | `capability_auth_flow_invalid` |
 | Device auth flow expired | `capability_auth_flow_expired` |
 | Device auth flow denied by user | `capability_auth_flow_denied` |
+| Callback URL malformed | `capability_auth_callback_invalid` |
+| Callback/code mismatch | `capability_auth_code_conflict` |
+| Callback state mismatch | `capability_auth_state_mismatch` |
+| Missing authorization code | `capability_auth_code_missing` |
 | Invalid input schema | `capability_invalid_input` |
 | Upstream/provider unavailable | `capability_backend_unavailable` |
 

specs/capability-auth.md

@@ -31,7 +31,7 @@ The device code flow works identically across all providers.
 1. Auth flows work on any deployment: headless server, SSH session, Telegram bot.
 2. No browser or public URL is required on the machine running Ash.
 3. Users complete consent on their own device — phone, laptop, or any browser.
-4. Existing `auth_begin`/`auth_complete` (authorization code) flows continue to work unchanged.
+4. Existing caller-facing `auth_begin`/`auth_complete` flows continue to work; auth completion normalization is centralized in host capability manager.
 5. Skills detect flow type from `auth_begin` response and adapt UX accordingly.
 
 ## Device Code Flow
@@ -325,7 +325,8 @@ Update `src/ash/integrations/skills/capabilities/google/SKILL.md` auth section t
 
 - `flow_type` defaults to `"authorization_code"` — existing bridges and skill flows keep working.
 - `auth_poll` is handled by the manager delegation wrapper: when the provider is missing or doesn't implement it, raises `CapabilityError("capability_invalid_input", "auth polling not supported by this provider")`.
-- Existing `auth_complete` with `code`/`callback_url` continues to work for redirect-based flows.
+- Existing caller-side `auth_complete` with `code`/`callback_url` continues to work for redirect-based flows.
+- Provider/bridge `auth_complete` consumes normalized `authorization_code` input; callback URL parsing is host-owned.
 - `SubprocessCapabilityProvider` handles missing `auth_poll` from bridges gracefully (bridge returns error, provider surfaces it).
 - New fields in `auth_begin` return dict (`flow_type`, `user_code`, `poll_interval_seconds`) are nullable/defaulted — callers that don't check them are unaffected.
 

src/ash/capabilities/__init__.py

@@ -21,6 +21,7 @@
 )
 from ash.capabilities.providers import (
     CapabilityAuthBeginResult,
+    CapabilityAuthCompleteInput,
     CapabilityAuthCompleteResult,
     CapabilityAuthPollResult,
     CapabilityCallContext,
@@ -41,6 +42,7 @@
     "CapabilityProvider",
     "CapabilityCallContext",
     "CapabilityAuthBeginResult",
+    "CapabilityAuthCompleteInput",
     "CapabilityAuthCompleteResult",
     "CapabilityAuthPollResult",
     "CapabilityAccount",

src/ash/capabilities/auth_normalization.py

@@ -0,0 +1,98 @@
+"""Central auth-complete input normalization for capability auth flows."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from urllib.parse import parse_qs, urlparse
+
+
+class AuthNormalizationError(ValueError):
+    """Normalization error with stable capability auth error code."""
+
+    def __init__(self, code: str, message: str) -> None:
+        super().__init__(message)
+        self.code = code
+
+
+@dataclass(slots=True)
+class NormalizedAuthCompletion:
+    """Canonical auth completion payload used by capability providers."""
+
+    authorization_code: str
+    raw_callback_url: str | None
+    state: str | None
+
+
+def normalize_auth_completion(
+    *,
+    callback_url: str | None,
+    code: str | None,
+    expected_state: str | None,
+) -> NormalizedAuthCompletion:
+    """Normalize callback URL / code inputs into one authorization code."""
+    normalized_code = _optional_text(code)
+    normalized_callback_url = _optional_text(callback_url)
+    callback_code: str | None = None
+    callback_state: str | None = None
+
+    if normalized_callback_url is not None:
+        callback_code, callback_state = _parse_callback_url(normalized_callback_url)
+
+    if (
+        normalized_code is not None
+        and callback_code is not None
+        and normalized_code != callback_code
+    ):
+        raise AuthNormalizationError(
+            "capability_auth_code_conflict",
+            "code does not match callback_url code",
+        )
+
+    if (
+        expected_state is not None
+        and callback_state is not None
+        and callback_state != expected_state
+    ):
+        raise AuthNormalizationError(
+            "capability_auth_state_mismatch",
+            "callback_url state does not match auth flow",
+        )
+
+    authorization_code = normalized_code or callback_code
+    if authorization_code is None:
+        raise AuthNormalizationError(
+            "capability_auth_code_missing",
+            "either code or callback_url with code is required",
+        )
+
+    return NormalizedAuthCompletion(
+        authorization_code=authorization_code,
+        raw_callback_url=normalized_callback_url,
+        state=callback_state,
+    )
+
+
+def _parse_callback_url(callback_url: str) -> tuple[str, str | None]:
+    parsed = urlparse(callback_url)
+    if not parsed.scheme or not parsed.netloc:
+        raise AuthNormalizationError(
+            "capability_auth_callback_invalid",
+            "callback_url is not a valid URL",
+        )
+
+    query = parse_qs(parsed.query)
+    code = _optional_text((query.get("code") or [None])[0])
+    if code is None:
+        raise AuthNormalizationError(
+            "capability_auth_code_missing",
+            "callback_url missing code query parameter",
+        )
+    state = _optional_text((query.get("state") or [None])[0])
+    return code, state
+
+
+def _optional_text(value: str | None) -> str | None:
+    if value is None:
+        return None
+    text = str(value).strip()
+    return text or None

src/ash/capabilities/manager.py

@@ -11,8 +11,13 @@
 from datetime import UTC, datetime, timedelta
 from typing import Any
 
+from ash.capabilities.auth_normalization import (
+    AuthNormalizationError,
+    normalize_auth_completion,
+)
 from ash.capabilities.providers import (
     CapabilityAuthBeginResult,
+    CapabilityAuthCompleteInput,
     CapabilityAuthCompleteResult,
     CapabilityAuthPollResult,
     CapabilityCallContext,
@@ -282,6 +287,7 @@ async def auth_begin(
                 expires_at=expires_at,
                 flow_state=dict(begin_result.flow_state),
                 flow_type=flow_type,
+                expected_callback_state=begin_result.expected_callback_state,
             )
 
         result: dict[str, Any] = {
@@ -331,11 +337,6 @@ async def auth_complete(
         normalized_source_display_name = _optional_text(source_display_name)
         normalized_callback_url = _optional_text(callback_url)
         normalized_code = _optional_text(code)
-        if not normalized_callback_url and not normalized_code:
-            raise CapabilityError(
-                "capability_invalid_input",
-                "either callback_url or code is required",
-            )
 
         async with self._lock:
             self._prune_expired_flows_locked()
@@ -353,6 +354,14 @@ async def auth_complete(
             _, provider_impl = self._get_definition_and_provider_locked(
                 flow.capability_id
             )
+        try:
+            normalized_completion = normalize_auth_completion(
+                callback_url=normalized_callback_url,
+                code=normalized_code,
+                expected_state=flow.expected_callback_state,
+            )
+        except AuthNormalizationError as e:
+            raise CapabilityError(e.code, str(e)) from e
 
         call_context = CapabilityCallContext(
             user_id=normalized_user_id,
@@ -368,8 +377,11 @@ async def auth_complete(
             provider_impl,
             capability_id=flow.capability_id,
             flow_state=dict(flow.flow_state),
-            callback_url=normalized_callback_url,
-            code=normalized_code,
+            completion=CapabilityAuthCompleteInput(
+                authorization_code=normalized_completion.authorization_code,
+                raw_callback_url=normalized_completion.raw_callback_url,
+                state=normalized_completion.state,
+            ),
             account_hint=flow.account_hint,
             context=call_context,
         )
@@ -659,6 +671,7 @@ async def _provider_auth_begin(
             flow_type=result.flow_type or "authorization_code",
             user_code=result.user_code,
             poll_interval_seconds=result.poll_interval_seconds,
+            expected_callback_state=result.expected_callback_state,
         )
 
     async def _provider_auth_poll(
@@ -686,8 +699,7 @@ async def _provider_auth_complete(
         *,
         capability_id: str,
         flow_state: dict[str, Any],
-        callback_url: str | None,
-        code: str | None,
+        completion: CapabilityAuthCompleteInput,
         account_hint: str | None,
         context: CapabilityCallContext,
     ) -> CapabilityAuthCompleteResult:
@@ -698,8 +710,7 @@ async def _provider_auth_complete(
         return await provider_impl.auth_complete(
             capability_id=capability_id,
             flow_state=flow_state,
-            callback_url=callback_url,
-            code=code,
+            completion=completion,
             context=context,
         )
 

src/ash/capabilities/providers/__init__.py

@@ -2,6 +2,7 @@
 
 from ash.capabilities.providers.base import (
     CapabilityAuthBeginResult,
+    CapabilityAuthCompleteInput,
     CapabilityAuthCompleteResult,
     CapabilityAuthPollResult,
     CapabilityCallContext,
@@ -11,6 +12,7 @@
 
 __all__ = [
     "CapabilityAuthBeginResult",
+    "CapabilityAuthCompleteInput",
     "CapabilityAuthCompleteResult",
     "CapabilityAuthPollResult",
     "CapabilityCallContext",

src/ash/capabilities/providers/base.py

@@ -36,6 +36,7 @@ class CapabilityAuthBeginResult:
     flow_type: str = "authorization_code"  # or "device_code"
     user_code: str | None = None
     poll_interval_seconds: int | None = None
+    expected_callback_state: str | None = None
 
 
 @dataclass(slots=True)
@@ -47,6 +48,15 @@ class CapabilityAuthCompleteResult:
     metadata: dict[str, Any] = field(default_factory=dict)
 
 
+@dataclass(slots=True)
+class CapabilityAuthCompleteInput:
+    """Normalized auth completion input for providers."""
+
+    authorization_code: str
+    raw_callback_url: str | None = None
+    state: str | None = None
+
+
 @dataclass(slots=True)
 class CapabilityAuthPollResult:
     """Provider response for device code auth polling."""
@@ -94,8 +104,7 @@ async def auth_complete(
         *,
         capability_id: str,
         flow_state: dict[str, Any],
-        callback_url: str | None,
-        code: str | None,
+        completion: CapabilityAuthCompleteInput,
         context: CapabilityCallContext,
     ) -> CapabilityAuthCompleteResult:
         """Complete auth flow and return linked account result."""

src/ash/capabilities/providers/subprocess.py

@@ -19,6 +19,7 @@
 
 from ash.capabilities.providers.base import (
     CapabilityAuthBeginResult,
+    CapabilityAuthCompleteInput,
     CapabilityAuthCompleteResult,
     CapabilityAuthPollResult,
     CapabilityCallContext,
@@ -121,6 +122,9 @@ async def auth_begin(
             flow_type=flow_type,
             user_code=user_code,
             poll_interval_seconds=poll_interval,
+            expected_callback_state=_optional_text(
+                result.get("expected_callback_state")
+            ),
         )
 
     async def auth_poll(
@@ -168,17 +172,17 @@ async def auth_complete(
         *,
         capability_id: str,
         flow_state: dict[str, Any],
-        callback_url: str | None,
-        code: str | None,
+        completion: CapabilityAuthCompleteInput,
         context: CapabilityCallContext,
     ) -> CapabilityAuthCompleteResult:
         result = await self._call_bridge(
             "auth_complete",
             {
                 "capability_id": capability_id,
                 "flow_state": dict(flow_state),
-                "callback_url": callback_url,
-                "code": code,
+                "authorization_code": completion.authorization_code,
+                "raw_callback_url": completion.raw_callback_url,
+                "state": completion.state,
                 "context_token": self._issue_context_token(context),
             },
         )
@@ -526,6 +530,13 @@ def _required_text(*, value: Any, code: str, message: str) -> str:
     return text
 
 
+def _optional_text(value: Any) -> str | None:
+    if value is None:
+        return None
+    text = str(value).strip()
+    return text or None
+
+
 def _capability_error(code: str, message: str) -> Exception:
     from ash.capabilities.manager import CapabilityError
 

src/ash/capabilities/types.py

@@ -44,6 +44,7 @@ class CapabilityAuthFlow:
     expires_at: datetime
     flow_state: dict[str, Any] = field(default_factory=dict)
     flow_type: str = "authorization_code"
+    expected_callback_state: str | None = None
 
 
 @dataclass(slots=True)

src/ash/skills/bundled/gog/scripts/gogcli_bridge.py

@@ -739,6 +739,7 @@ def _handle_auth_begin(params: dict[str, Any]) -> dict[str, Any]:
         "expires_at": _iso8601_utc(expires_epoch),
         "flow_state": flow_state,
         "flow_type": "authorization_code",
+        "expected_callback_state": state_param,
     }
 
 
@@ -986,9 +987,9 @@ def _handle_auth_complete(params: dict[str, Any]) -> dict[str, Any]:
         or "default"
     )
     code = _required_text(
-        params.get("code"),
+        params.get("authorization_code"),
         code="capability_invalid_input",
-        message="code is required for auth_complete",
+        message="authorization_code is required for auth_complete",
     )
     redirect_uri = (
         _optional_text(stored_flow.get("redirect_uri")) or AUTH_CODE_REDIRECT_URI