deploy

deploy

Concrete wiring of the multi-tenant datacrew resolver for deployment.

Builds a DatacrewResolver with production dependency impls

• verify_token: validate the dc_ JWT against datacrew.space JWKS (RS256)
• check_entitlement: GET /api/internal/entitlements/check (X-Webui-Secret)
• fetch_cred: GET /api/internal/domo-cred/{email}/{slug} (X-Webui-Secret)
• decrypt: envelope.decrypt with the VPS RSA private key (the real wiring)

The network deps are overridable so the wiring is unit-testable; defaults make real HTTP/JWKS calls and are exercised at deploy time (integration), not here.

build_datacrew_resolver

build_datacrew_resolver(
    *,
    rsa_private_key_pem: str,
    jwks_url: str = "https://datacrew.space/.well-known/jwks.json",
    issuer: str = "https://datacrew.space",
    audience: str = "https://domo.datacrew.space",
    internal_base_url: str = "https://datacrew.space",
    internal_secret: str = "",
    entitlement_resource: str = "crew-dcs",
    verify_token: Callable[[str | None], str] | None = None,
    check_entitlement: (
        Callable[[str], Awaitable[bool]] | None
    ) = None,
    fetch_cred: (
        Callable[[str, str], Awaitable[dict]] | None
    ) = None,
    validate: Callable[[Any], Awaitable[bool]] | None = None
) -> DatacrewResolver

Wire a production DatacrewResolver. Network deps default to real impls; pass overrides (verify_token/check_entitlement/fetch_cred) to unit-test.

Source code in src/crew_dcs/mcp_server/deploy.py

def build_datacrew_resolver(
    *,
    rsa_private_key_pem: str,
    jwks_url: str = "https://datacrew.space/.well-known/jwks.json",
    issuer: str = "https://datacrew.space",
    audience: str = "https://domo.datacrew.space",
    internal_base_url: str = "https://datacrew.space",
    internal_secret: str = "",
    entitlement_resource: str = "crew-dcs",
    verify_token: Callable[[str | None], str] | None = None,
    check_entitlement: Callable[[str], Awaitable[bool]] | None = None,
    fetch_cred: Callable[[str, str], Awaitable[dict]] | None = None,
    validate: Callable[[Any], Awaitable[bool]] | None = None,
) -> DatacrewResolver:
    """Wire a production DatacrewResolver. Network deps default to real impls;
    pass overrides (verify_token/check_entitlement/fetch_cred) to unit-test."""
    return DatacrewResolver(
        verify_token=verify_token or _default_verify_token(jwks_url, issuer, audience),
        check_entitlement=check_entitlement
        or _default_check_entitlement(
            internal_base_url, internal_secret, entitlement_resource
        ),
        fetch_cred=fetch_cred
        or _default_fetch_cred(internal_base_url, internal_secret),
        decrypt=lambda ciphertext: envelope.decrypt(ciphertext, rsa_private_key_pem),
        validate=validate,
    )

build_jwt_verifier

build_jwt_verifier(
    *, jwks_url: str, issuer: str, audience: str
)

A FastMCP JWTVerifier that tolerates the dc_ wire prefix.

MCP clients send Authorization: Bearer dc_<jwt>. The stock JWTVerifier would try to parse the literal dc_<jwt> (not a valid JWT) and 401 every request, so we strip the prefix before delegating to the real verification.

Source code in src/crew_dcs/mcp_server/deploy.py

def build_jwt_verifier(*, jwks_url: str, issuer: str, audience: str):
    """A FastMCP JWTVerifier that tolerates the `dc_` wire prefix.

    MCP clients send `Authorization: Bearer dc_<jwt>`. The stock JWTVerifier would
    try to parse the literal `dc_<jwt>` (not a valid JWT) and 401 every request, so
    we strip the prefix before delegating to the real verification.
    """
    from fastmcp.server.auth.providers.jwt import JWTVerifier

    class _DcPrefixJWTVerifier(JWTVerifier):
        async def verify_token(self, token):
            return await super().verify_token(strip_dc_prefix(token))

    return _DcPrefixJWTVerifier(jwks_uri=jwks_url, issuer=issuer, audience=audience)

build_validate_handler

build_validate_handler(
    *,
    rsa_private_key_pem: str,
    internal_base_url: str,
    internal_secret: str,
    fetch_cred: (
        Callable[[str, str], Awaitable[dict | None]] | None
    ) = None,
    who_am_i: (
        Callable[[str, str], Awaitable[Any]] | None
    ) = None
)

Build the async Starlette handler for POST /internal/validate.

Server-to-server: validates an ALREADY-STORED, envelope-encrypted Domo credential the website cannot decrypt (only this VPS holds the RSA private key). Auth is the shared X-Webui-Secret header — this route is NOT behind the dc_ JWTVerifier. The token is never returned, only a boolean verdict.

Network deps (fetch_cred/who_am_i) default to the real httpx impls and are injectable for unit tests.

Source code in src/crew_dcs/mcp_server/deploy.py

def build_validate_handler(
    *,
    rsa_private_key_pem: str,
    internal_base_url: str,
    internal_secret: str,
    fetch_cred: Callable[[str, str], Awaitable[dict | None]] | None = None,
    who_am_i: Callable[[str, str], Awaitable[Any]] | None = None,
):
    """Build the async Starlette handler for ``POST /internal/validate``.

    Server-to-server: validates an ALREADY-STORED, envelope-encrypted Domo
    credential the website cannot decrypt (only this VPS holds the RSA private
    key). Auth is the shared ``X-Webui-Secret`` header — this route is NOT behind
    the dc_ JWTVerifier. The token is never returned, only a boolean verdict.

    Network deps (``fetch_cred``/``who_am_i``) default to the real httpx impls
    and are injectable for unit tests.
    """
    from starlette.responses import JSONResponse

    fetch = fetch_cred or _default_fetch_cred_or_none(
        internal_base_url, internal_secret
    )
    probe = who_am_i or _default_who_am_i()

    async def handler(request):
        if request.headers.get("x-webui-secret") != internal_secret:
            return JSONResponse({"error": "unauthorized"}, status_code=401)

        body = await request.json()
        email = body["email"]
        slug = body["slug"]

        cred = await fetch(email, slug)
        if not cred:
            return JSONResponse({"ok": False, "reason": "not_found"})

        token = envelope.decrypt(cred["ciphertext"], rsa_private_key_pem)
        instance = cred["domo_instance"]

        try:
            res = await probe(instance, token)
        except Exception:  # noqa: BLE001
            return JSONResponse({"ok": False, "reason": "unreachable"})

        if res.status_code == 200:
            return JSONResponse({"ok": True})
        return JSONResponse({"ok": False, "reason": "rejected"})

    return handler

strip_dc_prefix

strip_dc_prefix(token: str | None) -> str | None

Return the bare JWT, dropping a leading dc_ if present. None stays None.

Source code in src/crew_dcs/mcp_server/deploy.py

def strip_dc_prefix(token: str | None) -> str | None:
    """Return the bare JWT, dropping a leading `dc_` if present. None stays None."""
    if token and token.startswith(DC_TOKEN_PREFIX):
        return token[len(DC_TOKEN_PREFIX) :]
    return token