lda.chat/lda-wf
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

docs: plan auth admin read surface

Browse Source
This commit is contained in:
lda
2026-06-06 11:21:50 +07:00 Verified
parent 645d9f3c5d
commit 58771e4423
3 changed files with 927 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/superpowers/plans/2026-06-06-auth-admin-read-slice-3.md
+903
View File
@@ -0,0 +1,903 @@
# Auth Admin Read Slice 3 Implementation Plan
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
**Goal:** Add a read-only auth admin surface that lists and inspects auth records without exposing secret payload values.
**Architecture:** Extend the existing neutral `WorkflowAdminApi` surface instead of creating a new top-level server field. Providers return safe auth summaries only: id, scheme, metadata, payload keys. MCP implements the provider from its existing store; JSON-RPC and CLI expose read-only methods. No save/delete/auth mutation and no provider-specific display promises in this slice.
**Tech Stack:** Python 3.14, dataclasses, protocols, Typer, JSON-RPC HTTP, pytest, ruff, basedpyright.
---
## Scope
Implement only:
- read-only auth summaries
- inspect one auth summary by auth ref
- JSON-RPC methods
- RPC client methods
- CLI commands under `wf admin auth`
Do not implement:
- secret payload output
- auth save/delete/update commands
- provider-specific display structs
- OAuth/secret-manager behavior
- changes to source registry mutation behavior
## Files
- Modify: `src/wf_api/admin.py`
- add `WorkflowAdminAuthProvider`
- add optional auth provider to `WorkflowAdminApi`
- add `list_auth_records` / `inspect_auth_record`
- Modify: `src/wf_api/surface.py`
- add auth methods to `WorkflowAdminSurface`
- Modify: `src/wf_api/__init__.py`
- export `WorkflowAdminAuthProvider`
- Modify: `src/wf_mcp/storage/store.py`
- add `list_auth_refs`
- Create: `src/wf_mcp/broker/service/auth_admin.py`
- `McpAuthAdminProvider`
- Modify: `src/wf_mcp/broker/service/upstream_transport.py`
- expose auth admin provider or list refs through store if needed
- Modify: `src/wf_mcp/broker/server.py`
- wire `WorkflowAdminApi(..., auth=...)`
- Modify: `src/wf_server/context.py`
- local/static admin uses no auth provider and reports unavailable
- Modify: `src/wf_transport_rpc_http/methods_admin.py`
- register `workflow.admin.auth.list` / `.inspect`
- Modify: `src/wf_transport_rpc_http/models.py`
- add `InspectAuthParams`
- Modify: `src/wf_transport_rpc_http/client_admin.py`
- add client methods
- Modify: `src/wf_cli/commands/admin.py`
- add auth sub-Typer
- Create: `src/wf_cli/commands/auth_admin.py`
- `wf admin auth list` / `inspect`
- Tests:
- `tests/wf_api/test_admin_api.py` or create if absent
- `tests/wf_mcp/service/test_auth_admin.py`
- `tests/wf_transport_rpc_http/test_admin_auth_rpc.py`
- `tests/wf_cli/test_auth_admin.py`
- Docs:
- `docs/current_roadmap.md`
- `docs/superpowers/specs/2026-06-06-auth-source-secrets-boundary.md`
## Task 1: Neutral admin auth surface
**Files:**
- Modify: `src/wf_api/admin.py`
- Modify: `src/wf_api/surface.py`
- Modify: `src/wf_api/__init__.py`
- Test: `tests/wf_api/test_admin_api.py`
- [ ] **Step 1: Add/extend wf_api admin tests**
If `tests/wf_api/test_admin_api.py` does not exist, create it with the imports
below. If it exists, append these tests.
```python
from __future__ import annotations
import pytest
from wf_api.admin import WorkflowAdminApi
class EmptyConnectionProvider:
def list_connections(self):
return []
def get_connection_statuses(self):
return []
class EmptyEventProvider:
def list_events(self):
return []
class AuthProvider:
def list_auth_records(self):
return [
{
"id": "github.work",
"scheme": "bearer",
"metadata": {"owner": "platform"},
"payload_keys": ["token"],
},
{
"id": "api.work",
"scheme": "headers",
"metadata": {},
"payload_keys": ["headers"],
},
]
def inspect_auth_record(self, auth_ref: str):
for record in self.list_auth_records():
if record["id"] == auth_ref:
return record
raise KeyError(auth_ref)
def _api(auth=None) -> WorkflowAdminApi:
return WorkflowAdminApi(
connections=EmptyConnectionProvider(),
events=EmptyEventProvider(),
auth=auth,
)
async def test_admin_lists_auth_records_sorted_without_payload_values() -> None:
payload = await _api(AuthProvider()).list_auth_records()
assert payload["total"] == 2
assert [record["id"] for record in payload["auth_records"]] == [
"api.work",
"github.work",
]
assert payload["auth_records"][0]["payload_keys"] == ["headers"]
assert "payload" not in payload["auth_records"][0]
async def test_admin_inspects_auth_record_without_payload_values() -> None:
payload = await _api(AuthProvider()).inspect_auth_record("github.work")
assert payload == {
"id": "github.work",
"scheme": "bearer",
"metadata": {"owner": "platform"},
"payload_keys": ["token"],
}
async def test_admin_auth_methods_report_unavailable_without_provider() -> None:
with pytest.raises(RuntimeError, match="auth admin is not available"):
await _api().list_auth_records()
with pytest.raises(RuntimeError, match="auth admin is not available"):
await _api().inspect_auth_record("github.work")
```
- [ ] **Step 2: Run tests to verify failure**
Run:
```bash
uv run pytest tests/wf_api/test_admin_api.py -q
```
Expected: fails because `WorkflowAdminApi` does not accept `auth` and auth methods do not exist.
- [ ] **Step 3: Add provider protocol and API methods**
Modify `src/wf_api/admin.py`.
Add protocol:
```python
class WorkflowAdminAuthProvider(Protocol):
"""Provides read-only auth inventory without secret payload values."""
def list_auth_records(self) -> Sequence[Mapping[str, Any] | object]: ...
def inspect_auth_record(self, auth_ref: str) -> Mapping[str, Any] | object: ...
```
Change `WorkflowAdminApi.__init__` signature:
```python
def __init__(
self,
*,
connections: WorkflowAdminConnectionProvider,
events: WorkflowAdminEventProvider,
auth: WorkflowAdminAuthProvider | None = None,
) -> None:
self.connections = connections
self.events = events
self.auth = auth
```
Add methods:
```python
async def list_auth_records(self) -> dict[str, Any]:
if self.auth is None:
raise RuntimeError("auth admin is not available for this target")
records = sorted(
(_payload(item) for item in self.auth.list_auth_records()),
key=lambda item: str(item.get("id", "")),
)
return {"auth_records": records, "total": len(records)}
async def inspect_auth_record(self, auth_ref: str) -> dict[str, Any]:
if self.auth is None:
raise RuntimeError("auth admin is not available for this target")
return _payload(self.auth.inspect_auth_record(auth_ref))
```
- [ ] **Step 4: Update surface protocol**
Modify `src/wf_api/surface.py`.
In `WorkflowAdminSurface`, add:
```python
async def list_auth_records(self) -> dict[str, Any]: ...
async def inspect_auth_record(self, auth_ref: str) -> dict[str, Any]: ...
```
If `Any` is not imported in that file, add:
```python
from typing import Any
```
- [ ] **Step 5: Export provider**
Modify `src/wf_api/__init__.py`.
Add to admin import:
```python
WorkflowAdminAuthProvider,
```
Add to `__all__`:
```python
"WorkflowAdminAuthProvider",
```
- [ ] **Step 6: Run focused tests**
Run:
```bash
uv run pytest tests/wf_api/test_admin_api.py tests/wf_api/test_import_direction.py -q
uv run ruff check src/wf_api/admin.py src/wf_api/surface.py src/wf_api/__init__.py tests/wf_api/test_admin_api.py
uv run basedpyright --level error src/wf_api tests/wf_api/test_admin_api.py
```
Expected: all pass.
## Task 2: MCP auth admin provider
**Files:**
- Modify: `src/wf_mcp/storage/store.py`
- Create: `src/wf_mcp/broker/service/auth_admin.py`
- Test: `tests/wf_mcp/service/test_auth_admin.py`
- [ ] **Step 1: Add provider tests**
Create `tests/wf_mcp/service/test_auth_admin.py`:
```python
from __future__ import annotations
from pathlib import Path
import pytest
from wf_mcp.broker.service.auth_admin import McpAuthAdminProvider
from wf_mcp.models import AuthRecord
from wf_mcp.storage import FileStore
def _store(tmp_path: Path) -> FileStore:
return FileStore(tmp_path)
def test_auth_admin_lists_safe_summaries_sorted(tmp_path: Path) -> None:
store = _store(tmp_path)
store.save_auth(
AuthRecord(
connection_id="github.work",
scheme="bearer",
payload={"token": "secret", "headers": {"Authorization": "Bearer secret"}},
)
)
store.save_auth(
AuthRecord(
connection_id="api.work",
scheme="headers",
payload={"headers": {"X-API-Key": "secret"}},
)
)
provider = McpAuthAdminProvider(store=store)
records = provider.list_auth_records()
assert records == [
{
"id": "api.work",
"scheme": "headers",
"metadata": {},
"payload_keys": ["headers"],
},
{
"id": "github.work",
"scheme": "bearer",
"metadata": {},
"payload_keys": ["headers", "token"],
},
]
def test_auth_admin_inspects_safe_summary(tmp_path: Path) -> None:
store = _store(tmp_path)
store.save_auth(
AuthRecord(
connection_id="github.work",
scheme="bearer",
payload={"token": "secret"},
)
)
provider = McpAuthAdminProvider(store=store)
assert provider.inspect_auth_record("github.work") == {
"id": "github.work",
"scheme": "bearer",
"metadata": {},
"payload_keys": ["token"],
}
def test_auth_admin_inspect_unknown_raises_key_error(tmp_path: Path) -> None:
provider = McpAuthAdminProvider(store=_store(tmp_path))
with pytest.raises(KeyError, match="unknown auth record"):
provider.inspect_auth_record("missing.auth")
```
- [ ] **Step 2: Run tests to verify failure**
Run:
```bash
uv run pytest tests/wf_mcp/service/test_auth_admin.py -q
```
Expected: fails because `McpAuthAdminProvider` does not exist and `FileStore` cannot list auth refs.
- [ ] **Step 3: Add `list_auth_refs` to store**
Modify `src/wf_mcp/storage/store.py`.
Add to `class Store`:
```python
def list_auth_refs(self) -> list[str]:
raise NotImplementedError
```
Add to `class FileStore`:
```python
def list_auth_refs(self) -> list[str]:
"""Return auth refs present in the local file auth store."""
return sorted(path.stem for path in self.auth_dir.glob("*.json"))
```
- [ ] **Step 4: Create MCP auth admin provider**
Create `src/wf_mcp/broker/service/auth_admin.py`:
```python
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
from wf_api import WorkflowAdminAuthProvider
from ...storage import Store
@dataclass(frozen=True, slots=True)
class McpAuthAdminProvider(WorkflowAdminAuthProvider):
"""Read-only auth inventory for MCP-backed workflow servers.
Summaries intentionally expose payload keys, not payload values. Concrete
auth variants can provide richer safe display later.
"""
store: Store
def list_auth_records(self) -> list[dict[str, Any]]:
return [
self.inspect_auth_record(auth_ref)
for auth_ref in sorted(self.store.list_auth_refs())
]
def inspect_auth_record(self, auth_ref: str) -> dict[str, Any]:
record = self.store.load_auth(auth_ref)
if record is None:
raise KeyError(f"unknown auth record {auth_ref!r}")
return {
"id": record.connection_id,
"scheme": record.scheme,
"metadata": {},
"payload_keys": sorted(str(key) for key in record.payload),
}
__all__ = ["McpAuthAdminProvider"]
```
- [ ] **Step 5: Run focused tests**
Run:
```bash
uv run pytest tests/wf_mcp/service/test_auth_admin.py tests/wf_mcp/test_store.py -q
uv run ruff check src/wf_mcp/storage/store.py src/wf_mcp/broker/service/auth_admin.py tests/wf_mcp/service/test_auth_admin.py
uv run basedpyright --level error src/wf_mcp/storage/store.py src/wf_mcp/broker/service/auth_admin.py tests/wf_mcp/service/test_auth_admin.py
```
Expected: all pass.
## Task 3: Wire MCP-backed server admin auth provider
**Files:**
- Modify: `src/wf_mcp/broker/server.py`
- Test: `tests/wf_mcp/test_mcp_workflow_server.py`
- [ ] **Step 1: Add server wiring test**
Append to `tests/wf_mcp/test_mcp_workflow_server.py`:
```python
async def test_workflow_server_from_service_exposes_auth_admin(tmp_path: Path) -> None:
from wf_mcp.models import AuthRecord
service = WfMcpService(store=FileStore(tmp_path))
service.save_auth(
AuthRecord(
connection_id="github.work",
scheme="bearer",
payload={"token": "secret"},
)
)
server = workflow_server_from_service(service)
payload = await server.admin.list_auth_records()
assert payload["auth_records"] == [
{
"id": "github.work",
"scheme": "bearer",
"metadata": {},
"payload_keys": ["token"],
}
]
```
If the file uses a helper for `WfMcpService`, follow its existing style but keep
`tmp_path`.
- [ ] **Step 2: Run test to verify failure**
Run:
```bash
uv run pytest tests/wf_mcp/test_mcp_workflow_server.py::test_workflow_server_from_service_exposes_auth_admin -q
```
Expected: fails because MCP-backed server does not wire an auth provider.
- [ ] **Step 3: Wire provider**
Modify `src/wf_mcp/broker/server.py`.
Add import:
```python
from .service.auth_admin import McpAuthAdminProvider
```
Find `WorkflowAdminApi(...)` construction in `workflow_server_from_service`.
Change it to pass:
```python
auth=McpAuthAdminProvider(store=service.store),
```
Do not add an auth provider to local/static server construction.
- [ ] **Step 4: Run focused tests**
Run:
```bash
uv run pytest tests/wf_mcp/test_mcp_workflow_server.py tests/wf_server/test_local_static_server.py -q
uv run ruff check src/wf_mcp/broker/server.py tests/wf_mcp/test_mcp_workflow_server.py
uv run basedpyright --level error src/wf_mcp/broker/server.py tests/wf_mcp/test_mcp_workflow_server.py
```
Expected: all pass. Local/static server should still report auth admin unavailable through `WorkflowAdminApi`.
## Task 4: JSON-RPC methods and client
**Files:**
- Modify: `src/wf_transport_rpc_http/models.py`
- Modify: `src/wf_transport_rpc_http/methods_admin.py`
- Modify: `src/wf_transport_rpc_http/client_admin.py`
- Test: `tests/wf_transport_rpc_http/test_admin_auth_rpc.py`
- [ ] **Step 1: Add RPC tests**
Create `tests/wf_transport_rpc_http/test_admin_auth_rpc.py`:
```python
from __future__ import annotations
import pytest
from wf_mcp.broker import WfMcpService
from wf_mcp.broker.server import workflow_server_from_service
from wf_mcp.models import AuthRecord
from wf_mcp.storage import FileStore
from wf_transport_rpc_http.app import create_rpc_app
from wf_transport_rpc_http.client import RpcWorkflowApiClient
@pytest.mark.anyio
async def test_rpc_lists_auth_records(tmp_path):
service = WfMcpService(store=FileStore(tmp_path))
service.save_auth(AuthRecord(connection_id="github.work", scheme="bearer", payload={"token": "secret"}))
app = create_rpc_app(workflow_server_from_service(service))
client = RpcWorkflowApiClient.from_asgi_app(app, url="http://test/rpc")
payload = await client.list_auth_records()
assert payload["auth_records"] == [
{
"id": "github.work",
"scheme": "bearer",
"metadata": {},
"payload_keys": ["token"],
}
]
@pytest.mark.anyio
async def test_rpc_inspects_auth_record(tmp_path):
service = WfMcpService(store=FileStore(tmp_path))
service.save_auth(AuthRecord(connection_id="github.work", scheme="bearer", payload={"token": "secret"}))
app = create_rpc_app(workflow_server_from_service(service))
client = RpcWorkflowApiClient.from_asgi_app(app, url="http://test/rpc")
payload = await client.inspect_auth_record("github.work")
assert payload["id"] == "github.work"
assert payload["payload_keys"] == ["token"]
assert "payload" not in payload
```
If existing RPC tests use `async def` without `pytest.mark.anyio`, follow the
existing local style instead.
- [ ] **Step 2: Run tests to verify failure**
Run:
```bash
uv run pytest tests/wf_transport_rpc_http/test_admin_auth_rpc.py -q
```
Expected: fails because RPC methods/client methods do not exist.
- [ ] **Step 3: Add params model**
Modify `src/wf_transport_rpc_http/models.py`.
Add:
```python
class InspectAuthParams(RpcBaseModel):
auth_ref: str = Field(min_length=1)
```
Use the same base model and `Field` import already used in this file.
- [ ] **Step 4: Register RPC methods**
Modify `src/wf_transport_rpc_http/methods_admin.py`.
Add import:
```python
from .models import AdminEmptyParams, InspectAuthParams
```
Add methods inside `register_methods`:
```python
@entrypoint.method(name="workflow.admin.auth.list", errors=[WorkflowRpcError])
async def workflow_admin_auth_list(
params: AdminEmptyParams = Body(default_factory=AdminEmptyParams),
) -> dict[str, Any]:
try:
return await server.admin.list_auth_records()
except (ValueError, KeyError, LookupError, FileNotFoundError, RuntimeError) as exc:
raise_workflow_rpc_error(exc)
@entrypoint.method(name="workflow.admin.auth.inspect", errors=[WorkflowRpcError])
async def workflow_admin_auth_inspect(
params: InspectAuthParams,
) -> dict[str, Any]:
try:
return await server.admin.inspect_auth_record(params.auth_ref)
except (ValueError, KeyError, LookupError, FileNotFoundError, RuntimeError) as exc:
raise_workflow_rpc_error(exc)
```
If ruff flags line length on the exception tuple, wrap it like existing methods.
- [ ] **Step 5: Add client methods**
Modify `src/wf_transport_rpc_http/client_admin.py`.
Add:
```python
async def list_auth_records(self) -> dict[str, Any]:
return await self._call("workflow.admin.auth.list", {})
async def inspect_auth_record(self, auth_ref: str) -> dict[str, Any]:
return await self._call(
"workflow.admin.auth.inspect",
{"auth_ref": auth_ref},
)
```
- [ ] **Step 6: Run focused tests**
Run:
```bash
uv run pytest tests/wf_transport_rpc_http/test_admin_auth_rpc.py -q
uv run ruff check src/wf_transport_rpc_http/models.py src/wf_transport_rpc_http/methods_admin.py src/wf_transport_rpc_http/client_admin.py tests/wf_transport_rpc_http/test_admin_auth_rpc.py
uv run basedpyright --level error src/wf_transport_rpc_http/models.py src/wf_transport_rpc_http/methods_admin.py src/wf_transport_rpc_http/client_admin.py tests/wf_transport_rpc_http/test_admin_auth_rpc.py
```
Expected: all pass.
## Task 5: CLI commands
**Files:**
- Create: `src/wf_cli/commands/auth_admin.py`
- Modify: `src/wf_cli/commands/admin.py`
- Test: `tests/wf_cli/test_auth_admin.py`
- [ ] **Step 1: Add CLI tests**
Create `tests/wf_cli/test_auth_admin.py`:
```python
from __future__ import annotations
from typer.testing import CliRunner
from wf_cli.app import app
class FakeAdmin:
async def list_auth_records(self):
return {
"auth_records": [
{
"id": "github.work",
"scheme": "bearer",
"metadata": {},
"payload_keys": ["token"],
}
],
"total": 1,
}
async def inspect_auth_record(self, auth_ref: str):
return {
"id": auth_ref,
"scheme": "bearer",
"metadata": {},
"payload_keys": ["token"],
}
class FakeContext:
admin = FakeAdmin()
verbose = False
def test_wf_admin_auth_list(monkeypatch) -> None:
monkeypatch.setattr(
"wf_cli.commands.auth_admin.load_cli_context_from_typer",
lambda ctx: FakeContext(),
)
result = CliRunner().invoke(app, ["admin", "auth", "list"])
assert result.exit_code == 0
assert "github.work" in result.stdout
assert "secret" not in result.stdout
def test_wf_admin_auth_inspect(monkeypatch) -> None:
monkeypatch.setattr(
"wf_cli.commands.auth_admin.load_cli_context_from_typer",
lambda ctx: FakeContext(),
)
result = CliRunner().invoke(app, ["admin", "auth", "inspect", "github.work"])
assert result.exit_code == 0
assert "github.work" in result.stdout
assert "payload_keys" in result.stdout
assert "secret" not in result.stdout
```
- [ ] **Step 2: Run tests to verify failure**
Run:
```bash
uv run pytest tests/wf_cli/test_auth_admin.py -q
```
Expected: fails because `wf admin auth` commands do not exist.
- [ ] **Step 3: Create CLI command module**
Create `src/wf_cli/commands/auth_admin.py`:
```python
from __future__ import annotations
from typing import Annotated
import typer
from wf_cli.context import load_cli_context_from_typer
from wf_cli.formats import ListOutputFormat, emit_detail_payload, emit_list_payload
from wf_cli.remote_errors import run_cli_operation
app = typer.Typer(
name="auth",
help="Read auth record status without exposing secret payload values.",
no_args_is_help=True,
)
@app.command("list")
def list_auth_records(
ctx: typer.Context,
output_format: Annotated[
ListOutputFormat, typer.Option("--format", help="Output format.")
] = ListOutputFormat.JSON,
) -> None:
"""List auth records known to the target."""
context = load_cli_context_from_typer(ctx)
payload = run_cli_operation(context, context.admin.list_auth_records())
emit_list_payload(
payload,
collection_key="auth_records",
output_format=output_format,
id_field="id",
summary_fields=("scheme", "payload_keys"),
)
@app.command("inspect")
def inspect_auth_record(
ctx: typer.Context,
auth_ref: Annotated[str, typer.Argument(help="Auth record id/ref.")],
) -> None:
"""Inspect one auth record summary without secret payload values."""
context = load_cli_context_from_typer(ctx)
payload = run_cli_operation(
context,
context.admin.inspect_auth_record(auth_ref),
)
emit_detail_payload(payload)
```
If `emit_detail_payload` has a different name in `wf_cli.formats`, inspect the
file and use the existing detail emitter.
- [ ] **Step 4: Register subcommand**
Modify `src/wf_cli/commands/admin.py`.
Change:
```python
from . import source_registry
```
to:
```python
from . import auth_admin, source_registry
```
Add after registry registration:
```python
app.add_typer(auth_admin.app, name="auth")
```
- [ ] **Step 5: Run focused tests**
Run:
```bash
uv run pytest tests/wf_cli/test_auth_admin.py -q
uv run ruff check src/wf_cli/commands/auth_admin.py src/wf_cli/commands/admin.py tests/wf_cli/test_auth_admin.py
uv run basedpyright --level error src/wf_cli/commands/auth_admin.py src/wf_cli/commands/admin.py tests/wf_cli/test_auth_admin.py
```
Expected: all pass.
## Task 6: Docs and final verification
**Files:**
- Modify: `docs/current_roadmap.md`
- Modify: `docs/superpowers/specs/2026-06-06-auth-source-secrets-boundary.md`
- [ ] **Step 1: Update spec status**
In `docs/superpowers/specs/2026-06-06-auth-source-secrets-boundary.md`, update
the `## Status` section to:
```markdown
Slice 1 implements the neutral auth record/store protocol and MCP compatibility
bridge. Slice 2 surfaces missing explicit auth refs through live source
diagnostics and source registry apply summaries. Slice 3 exposes read-only auth
admin summaries without secret payload values. Auth mutation surfaces and
provider-specific auth unions are future slices.
```
- [ ] **Step 2: Update roadmap**
In `docs/current_roadmap.md`, under the auth/source secrets boundary bullet,
append:
```markdown
Third implementation slice complete: read-only auth admin summaries are
available through MCP-backed server admin, JSON-RPC, and CLI. Summaries show
ids, schemes, metadata, and payload keys only; secret payload values remain
hidden.
```
- [ ] **Step 3: Run final verification**
Run:
```bash
uv run pytest tests/wf_api/test_admin_api.py tests/wf_mcp/service/test_auth_admin.py tests/wf_mcp/test_mcp_workflow_server.py tests/wf_transport_rpc_http/test_admin_auth_rpc.py tests/wf_cli/test_auth_admin.py -q
uv run ruff check src/wf_api/admin.py src/wf_api/surface.py src/wf_api/__init__.py src/wf_mcp/storage/store.py src/wf_mcp/broker/service/auth_admin.py src/wf_mcp/broker/server.py src/wf_transport_rpc_http/models.py src/wf_transport_rpc_http/methods_admin.py src/wf_transport_rpc_http/client_admin.py src/wf_cli/commands/auth_admin.py src/wf_cli/commands/admin.py tests/wf_api/test_admin_api.py tests/wf_mcp/service/test_auth_admin.py tests/wf_transport_rpc_http/test_admin_auth_rpc.py tests/wf_cli/test_auth_admin.py
uv run basedpyright --level error src/wf_api src/wf_mcp/broker/service/auth_admin.py src/wf_mcp/storage/store.py src/wf_transport_rpc_http src/wf_cli/commands/auth_admin.py tests/wf_api/test_admin_api.py tests/wf_mcp/service/test_auth_admin.py tests/wf_transport_rpc_http/test_admin_auth_rpc.py tests/wf_cli/test_auth_admin.py
```
Expected: all pass.
- [ ] **Step 4: Final report**
Report:
- files changed
- verification output
- final auth summary shape
- confirmation that no secret payload values are returned
- deviations from this plan
docs/superpowers/plans/2026-06-06-auth-diagnostics-slice-2.md
-552
View File
@@ -1,552 +0,0 @@
# Auth Diagnostics Slice 2 Implementation Plan
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
**Goal:** Surface missing auth references as explicit diagnostics in live source checks and source-registry apply summaries.
**Architecture:** Keep auth diagnostics MCP-provider-specific for now because current source/runtime auth resolution is implemented in `wf_mcp`. Add a small helper in `wf_mcp.auth` that inspects a `ConnectionConfig` and an auth lookup function, then reuse it from `UpstreamTransportService.deployment_diagnostics()` and `SourceRegistryAdminProvider.apply_registry_changes()`. Do not add auth admin CRUD, do not change auth storage, and do not make `wf_api` understand MCP credential payloads.
**Tech Stack:** Python 3.14, dataclasses, pytest, ruff, basedpyright.
---
## Scope
Implement only:
- `auth_not_found` diagnostic helper for connections with string `metadata["auth_ref"]`.
- live source diagnostics before upstream liveness probe.
- source registry apply summary field: `auth_diagnostics`.
- docs status update.
Do not implement:
- auth admin list/save/delete commands
- deployment static validation changes
- OAuth/secret-manager behavior
- source registry mutation rejection on missing auth
- any neutral `wf_api` diagnostic model changes
## Files
- Modify: `src/wf_mcp/auth.py`
- add `auth_ref_for_connection`
- add `auth_missing_diagnostic`
- add `connection_auth_diagnostic`
- Modify: `src/wf_mcp/broker/service/upstream_transport.py`
- use helper in `deployment_diagnostics`
- Modify: `src/wf_mcp/broker/service/source_registry_admin.py`
- accept optional auth loader
- return `auth_diagnostics` from `apply_registry_changes`
- Modify: `src/wf_mcp/broker/server.py`
- wire source-registry admin provider to upstream auth loader if needed
- Test: `tests/wf_mcp/test_auth.py`
- helper tests
- Test: `tests/wf_mcp/service/test_upstream_transport.py`
- live diagnostic for missing `auth_ref`
- Test: `tests/wf_mcp/service/test_source_registry_admin.py`
- apply summary includes auth diagnostics
- Docs: `docs/current_roadmap.md`
- Docs: `docs/superpowers/specs/2026-06-06-auth-source-secrets-boundary.md`
## Task 1: MCP auth diagnostic helper
**Files:**
- Modify: `src/wf_mcp/auth.py`
- Modify: `tests/wf_mcp/test_auth.py`
- [ ] **Step 1: Add helper tests**
Append to `tests/wf_mcp/test_auth.py`:
```python
from wf_artifacts import DiagnosticSeverity
from wf_mcp.auth import (
auth_ref_for_connection,
connection_auth_diagnostic,
)
from wf_mcp.models import ConnectionConfig
def test_auth_ref_for_connection_returns_string_only() -> None:
assert (
auth_ref_for_connection(
ConnectionConfig(
id="github.work",
server="github",
account="work",
metadata={"auth_ref": "github.creds"},
)
)
== "github.creds"
)
assert (
auth_ref_for_connection(
ConnectionConfig(
id="github.work",
server="github",
account="work",
metadata={"auth_ref": 123},
)
)
is None
)
def test_connection_auth_diagnostic_reports_missing_auth_ref() -> None:
connection = ConnectionConfig(
id="github.work",
server="github",
account="work",
metadata={"auth_ref": "github.creds"},
)
diagnostic = connection_auth_diagnostic(
connection,
load_auth=lambda auth_ref: None,
logical_ref="github",
)
assert diagnostic is not None
assert diagnostic.severity == DiagnosticSeverity.ERROR
assert diagnostic.code == "auth_not_found"
assert diagnostic.logical_ref == "github"
assert diagnostic.bound_source == "github.work"
assert "github.creds" in diagnostic.message
assert "Add an auth record" in diagnostic.repair_hint
def test_connection_auth_diagnostic_ignores_absent_or_present_auth_ref() -> None:
no_ref = ConnectionConfig(id="github.work", server="github", account="work")
with_ref = ConnectionConfig(
id="github.work",
server="github",
account="work",
metadata={"auth_ref": "github.creds"},
)
auth = McpAuthRecord(
connection_id="github.creds",
scheme="bearer",
payload={"token": "secret"},
)
assert (
connection_auth_diagnostic(
no_ref,
load_auth=lambda auth_ref: None,
logical_ref="github",
)
is None
)
assert (
connection_auth_diagnostic(
with_ref,
load_auth=lambda auth_ref: auth,
logical_ref="github",
)
is None
)
```
- [ ] **Step 2: Run tests to verify failure**
Run:
```bash
uv run pytest tests/wf_mcp/test_auth.py -q
```
Expected: fails because `auth_ref_for_connection` and `connection_auth_diagnostic` do not exist.
- [ ] **Step 3: Implement helper functions**
Modify `src/wf_mcp/auth.py`.
Add imports:
```python
from collections.abc import Callable
from wf_artifacts import DependencyDiagnostic, DiagnosticSeverity
from .models import ConnectionConfig
```
Add functions before `__all__`:
```python
def auth_ref_for_connection(connection: ConnectionConfig) -> str | None:
"""Return the explicit auth ref for one source connection, if present."""
auth_ref = connection.metadata.get("auth_ref")
return auth_ref if isinstance(auth_ref, str) else None
def auth_missing_diagnostic(
*,
auth_ref: str,
source_id: str,
logical_ref: str | None = None,
) -> DependencyDiagnostic:
"""Build a stable diagnostic without including secret payload data."""
return DependencyDiagnostic(
severity=DiagnosticSeverity.ERROR,
code="auth_not_found",
logical_ref=logical_ref,
bound_source=source_id,
message=(
f"Source {source_id!r} references auth record {auth_ref!r}, "
"but no auth record was found."
),
repair_hint=(
"Add an auth record for this auth_ref, update the source auth_ref, "
"or bind the deployment to a source that does not require it."
),
)
def connection_auth_diagnostic(
connection: ConnectionConfig,
*,
load_auth: Callable[[str], McpAuthRecord | None],
logical_ref: str | None = None,
) -> DependencyDiagnostic | None:
"""Return an auth diagnostic for explicit auth_ref misses.
Connections without explicit auth_ref keep legacy no-auth behavior. This
makes the new auth boundary observable without treating every unauthenticated
MCP source as an error.
"""
auth_ref = auth_ref_for_connection(connection)
if auth_ref is None:
return None
if load_auth(auth_ref) is not None:
return None
return auth_missing_diagnostic(
auth_ref=auth_ref,
source_id=connection.id,
logical_ref=logical_ref,
)
```
Add to `__all__`:
```python
"auth_missing_diagnostic",
"auth_ref_for_connection",
"connection_auth_diagnostic",
```
- [ ] **Step 4: Run focused tests**
Run:
```bash
uv run pytest tests/wf_mcp/test_auth.py -q
uv run ruff check src/wf_mcp/auth.py tests/wf_mcp/test_auth.py
uv run basedpyright --level error src/wf_mcp/auth.py tests/wf_mcp/test_auth.py
```
Expected: all pass.
## Task 2: Live source auth diagnostics
**Files:**
- Modify: `src/wf_mcp/broker/service/upstream_transport.py`
- Modify: `tests/wf_mcp/service/test_upstream_transport.py`
- [ ] **Step 1: Add live diagnostic test**
Append to `tests/wf_mcp/service/test_upstream_transport.py`:
```python
async def test_upstream_transport_live_diagnostics_report_missing_auth_ref(
tmp_path: Path,
) -> None:
events: list[McpEvent] = []
store = FileStore(tmp_path)
connections = ConnectionRegistry()
connection = ConnectionConfig(
id="github.work",
server="demo",
account="work",
metadata={"auth_ref": "github.creds"},
)
connections.register(connection)
transport = UpstreamTransportService(store=store, event_sink=events.append)
transport.register_adapter("demo", FakeAdapter())
source_catalog = SourceCatalogService(
store=store,
connection_lookup=connections.get,
connection_list_enabled=connections.list_enabled,
connection_list_all=connections.list_all,
tool_executor_for=transport.tool_executor_for,
load_auth=transport.load_connection_auth,
emit_event=events.append,
)
source_catalog.register_capability_source(
CapabilitySource(
id="github.work",
kind="connection",
permissions=SourcePermissions(calls_upstream=True),
capabilities=CapabilityBuckets(),
)
)
artifact = echo_artifact()
deployment = WorkflowDeployment(
id="echo.personal",
artifact_id="echo",
artifact_version=1,
bindings=[{"logical_source": "demo", "concrete_source": "github.work"}],
)
diagnostics = await transport.deployment_diagnostics(
deployment=deployment,
artifacts=[artifact],
source_catalog=source_catalog,
)
assert diagnostics[0].code == "auth_not_found"
assert diagnostics[0].bound_source == "github.work"
assert "github.creds" in diagnostics[0].message
```
- [ ] **Step 2: Run test to verify failure**
Run:
```bash
uv run pytest tests/wf_mcp/service/test_upstream_transport.py::test_upstream_transport_live_diagnostics_report_missing_auth_ref -q
```
Expected: fails because live diagnostics currently probe with `auth=None` instead of reporting `auth_not_found`.
- [ ] **Step 3: Add diagnostic before liveness probe**
Modify `src/wf_mcp/broker/service/upstream_transport.py`.
Add import:
```python
from wf_mcp.auth import connection_auth_diagnostic
```
Inside `deployment_diagnostics`, after `connection = source_catalog.connection_lookup(source_id)` succeeds and before `adapter = require_adapter(...)`, add:
```python
auth_diagnostic = connection_auth_diagnostic(
connection,
load_auth=self.load_auth,
logical_ref=logical_ref,
)
if auth_diagnostic is not None:
diagnostics.append(auth_diagnostic)
continue
```
Use `self.load_auth` here intentionally: it loads by auth ref. Do not use
`load_connection_auth`, because that would convert a missing explicit auth ref
into `None` and lose the diagnostic reason.
- [ ] **Step 4: Run focused tests**
Run:
```bash
uv run pytest tests/wf_mcp/service/test_upstream_transport.py -q
uv run ruff check src/wf_mcp/broker/service/upstream_transport.py tests/wf_mcp/service/test_upstream_transport.py
uv run basedpyright --level error src/wf_mcp/broker/service/upstream_transport.py tests/wf_mcp/service/test_upstream_transport.py
```
Expected: all pass.
## Task 3: Source registry apply auth diagnostics
**Files:**
- Modify: `src/wf_mcp/broker/service/source_registry_admin.py`
- Modify: `src/wf_mcp/broker/server.py`
- Modify: `tests/wf_mcp/service/test_source_registry_admin.py`
- [ ] **Step 1: Add apply summary test**
Append to `tests/wf_mcp/service/test_source_registry_admin.py`:
```python
def test_source_registry_apply_reports_missing_auth_ref(tmp_path: Path) -> None:
entry = McpSourceRegistryEntry(
id="github.work",
provider="github",
account="work",
auth_ref="github.creds",
transport=StdioSourceTransport(command="npx"),
)
provider, connection_service, _source_catalog = _apply_provider(
tmp_path,
registry_sources=[entry],
)
provider.load_auth = lambda auth_ref: None
payload = provider.apply_registry_changes()
assert payload["applied"] is True
assert payload["registered"] == ["github.work"]
assert connection_service.get("github.work").metadata["auth_ref"] == "github.creds"
assert payload["auth_diagnostics"] == [
{
"severity": "error",
"code": "auth_not_found",
"logical_ref": None,
"bound_source": "github.work",
"message": (
"Source 'github.work' references auth record 'github.creds', "
"but no auth record was found."
),
"repair_hint": (
"Add an auth record for this auth_ref, update the source auth_ref, "
"or bind the deployment to a source that does not require it."
),
}
]
```
If `DependencyDiagnostic.model_dump(mode="json")` uses enum values differently
in this repo, assert individual fields instead:
```python
diagnostic = payload["auth_diagnostics"][0]
assert diagnostic["code"] == "auth_not_found"
assert diagnostic["bound_source"] == "github.work"
assert "github.creds" in diagnostic["message"]
```
- [ ] **Step 2: Run test to verify failure**
Run:
```bash
uv run pytest tests/wf_mcp/service/test_source_registry_admin.py::test_source_registry_apply_reports_missing_auth_ref -q
```
Expected: fails because `SourceRegistryAdminProvider` has no `load_auth` field and apply summary has no `auth_diagnostics`.
- [ ] **Step 3: Add auth loader to provider**
Modify `src/wf_mcp/broker/service/source_registry_admin.py`.
Add imports:
```python
from ...auth import connection_auth_diagnostic
from ...models import AuthRecord
```
Add field to `SourceRegistryAdminProvider`:
```python
load_auth: Callable[[str], AuthRecord | None] | None = None
```
- [ ] **Step 4: Add diagnostics to apply summary**
In `apply_registry_changes`, after computing `after`, add:
```python
auth_diagnostics = []
if self.load_auth is not None:
for source_id in sorted(after):
diagnostic = connection_auth_diagnostic(
after[source_id],
load_auth=self.load_auth,
)
if diagnostic is not None:
auth_diagnostics.append(diagnostic.model_dump(mode="json"))
```
Then add to the returned dict:
```python
"auth_diagnostics": auth_diagnostics,
```
Do not reject apply when auth is missing. Apply reconciles desired source state;
auth diagnostics tell the operator why later live calls may fail.
- [ ] **Step 5: Wire runtime provider construction**
Modify `src/wf_mcp/broker/server.py`.
Find where `SourceRegistryAdminProvider(...)` is constructed for MCP-backed
`WorkflowServer`. Add:
```python
load_auth=service.upstream.load_auth,
```
If the file constructs the provider through a helper, pass the loader through
that helper. Do not change local/static server behavior; local/static still has
no source registry admin provider.
- [ ] **Step 6: Run focused tests**
Run:
```bash
uv run pytest tests/wf_mcp/service/test_source_registry_admin.py tests/wf_transport_rpc_http/test_mcp_backed_server_rpc.py -q
uv run ruff check src/wf_mcp/broker/service/source_registry_admin.py src/wf_mcp/broker/server.py tests/wf_mcp/service/test_source_registry_admin.py
uv run basedpyright --level error src/wf_mcp/broker/service/source_registry_admin.py src/wf_mcp/broker/server.py tests/wf_mcp/service/test_source_registry_admin.py
```
Expected: all pass.
## Task 4: Docs and verification
**Files:**
- Modify: `docs/current_roadmap.md`
- Modify: `docs/superpowers/specs/2026-06-06-auth-source-secrets-boundary.md`
- [ ] **Step 1: Update spec status**
In `docs/superpowers/specs/2026-06-06-auth-source-secrets-boundary.md`, update
the `## Status` section to:
```markdown
Slice 1 implements the neutral auth record/store protocol and MCP compatibility
bridge. Slice 2 surfaces missing explicit auth refs through live source
diagnostics and source registry apply summaries. Auth admin surfaces and
provider-specific auth unions are future slices.
```
- [ ] **Step 2: Update roadmap**
In `docs/current_roadmap.md`, under the auth/source secrets boundary bullet,
append:
```markdown
Second implementation slice complete: missing explicit auth refs now surface
as `auth_not_found` diagnostics in live source checks and source registry
apply summaries.
```
- [ ] **Step 3: Run final verification**
Run:
```bash
uv run pytest tests/wf_mcp/test_auth.py tests/wf_mcp/service/test_upstream_transport.py tests/wf_mcp/service/test_source_registry_admin.py tests/wf_transport_rpc_http/test_mcp_backed_server_rpc.py -q
uv run ruff check src/wf_mcp/auth.py src/wf_mcp/broker/service/upstream_transport.py src/wf_mcp/broker/service/source_registry_admin.py src/wf_mcp/broker/server.py tests/wf_mcp/test_auth.py tests/wf_mcp/service/test_upstream_transport.py tests/wf_mcp/service/test_source_registry_admin.py
uv run basedpyright --level error src/wf_mcp/auth.py src/wf_mcp/broker/service/upstream_transport.py src/wf_mcp/broker/service/source_registry_admin.py src/wf_mcp/broker/server.py tests/wf_mcp/test_auth.py tests/wf_mcp/service/test_upstream_transport.py tests/wf_mcp/service/test_source_registry_admin.py
```
Expected: all pass.
- [ ] **Step 4: Final report**
Report:
- files changed
- verification output
- final shape of `auth_diagnostics`
- any deviations from the plan
docs/superpowers/specs/2026-06-06-auth-source-secrets-boundary.md
+24
View File
@@ -164,6 +164,30 @@ Live source checks and source registry apply should prefer diagnostics over late
adapter failures. Runtime invocation can still fail if the upstream server
requires auth but does not declare that requirement.
## Read-Only Display
Read-only admin surfaces may show that an auth record exists, but they must not
promise provider-specific display until auth records are concrete variants.
For the current `scheme + payload` bridge, safe display should stay intentionally
minimal:
- `id`
- `scheme`
- `metadata`
- `payload_keys`
Do not expose payload values. Do not promise token hints, OAuth subjects,
expiry, scopes, header names, or environment-variable names as the stable
neutral contract yet. Once auth records become a discriminated union, each
variant can own a richer safe display method:
- bearer auth can show token presence or a redacted hint
- headers auth can show safe header names
- env auth can show safe environment variable names
- OAuth auth can show subject, expiry, scopes, and refreshability
- opaque auth can stay limited to scheme and payload keys
## Store Shape
The current filesystem store can remain: