Authentication

Overview

The connector does not log users in. Your application does, and tells the connector the role of the user on every request through the check_authentication callback. The role then selects the access rules.

Every request is authenticated on its own, so one connector serves different users with different roles at the same time.

Request pipeline

1. onlyPOST guard
2. GET /ping answers here
3. CORS (preflight requests end here)
4. resolve_sources (multi-tenant)
5. check_authentication → role
6. Parameters are parsed
7. Access check of the action for the role
8. Action

The callback

type AuthCallback = Callable[[Request], str | Awaitable[str]]

• Receives the Starlette Request (headers, cookies, request.session, request.state...).
• Returns the role as a string. Plain functions and coroutine functions both work.
• Raise HttpError to reject the request with a status of your choice (401, 403...). Any other exception becomes 500; so does a non-string result.
• Without the callback every request gets defaultRole (guest).

from starlette.requests import Request

from jcpy import create_app


async def check_authentication(request: Request) -> str:
    token = request.headers.get("authorization")
    if token is None:
        return "guest"
    user = await users.by_token(token)  # your code
    return user.role


app = create_app("config.json", check_authentication=check_authentication)

Role only

The callback returns a role, not a user. Per-user folders are made with dynamic sources, which also receive the request.

Cookie

The role comes from a cookie. The client can set any cookie, so use this only where the client is trusted, or keep the role server side (sessions, below).

def role_from_cookie(request: Request) -> str:
    """Read the role of the user from the ``userRole`` cookie.

    Args:
        request: Incoming request.

    Returns:
        Cookie value, ``guest`` when the cookie is missing.
    """
    return request.cookies.get(ROLE_COOKIE) or "guest"


def build_app() -> FastAPI:
    """Build the application.

    Returns:
        Connector application authenticating by cookie.
    """
    return create_app(CONFIG, check_authentication=role_from_cookie)

curl "http://localhost:8081/?action=permissions&source=uploads"
curl -H "Cookie: userRole=editor" "http://localhost:8081/?action=permissions&source=uploads"

Full program: examples/cookie_auth.py.

JWT

The token is verified (signature, expiry, required claims) with PyJWT; a bad token is answered with 401.

def make_token(
    role: str, username: str, lifetime: int = TOKEN_LIFETIME
) -> str:
    """Issue a token, as a login endpoint would.

    Args:
        role: Role of the user.
        username: Name of the user.
        lifetime: Seconds until the token expires.

    Returns:
        Signed token.
    """
    payload = {
        "role": role,
        "username": username,
        "exp": int(time.time()) + lifetime,
    }
    return jwt.encode(payload, SECRET, algorithm=ALGORITHM)


def role_from_jwt(request: Request) -> str:
    """Verify the bearer token and read the role from it.

    Args:
        request: Incoming request.

    Returns:
        ``role`` claim of the token, ``guest`` without a token.

    Raises:
        HttpError: ``401`` for a malformed, forged or expired token.
    """
    header = request.headers.get("authorization")
    if not header:
        return "guest"
    scheme, _, token = header.partition(" ")
    unauthorized = HttpError(
        HTTPStatus.UNAUTHORIZED, "Invalid or expired token"
    )
    if scheme.lower() != "bearer" or not token:
        raise unauthorized
    try:
        payload = jwt.decode(
            token,
            SECRET,
            algorithms=[ALGORITHM],
            options={"require": ["exp", "role"]},
        )
    except jwt.InvalidTokenError as error:
        raise unauthorized from error
    role = payload["role"]
    if not isinstance(role, str):
        raise unauthorized
    return role


def build_app() -> FastAPI:
    """Build the application.

    Returns:
        Connector application authenticating by JWT.
    """
    return create_app(CONFIG, check_authentication=role_from_jwt)

curl -H "Authorization: Bearer $TOKEN" \
  "http://localhost:8081/?action=permissions&source=uploads"

Full program: examples/jwt_auth.py.

Session

The role lives in a session signed by the server, set by your login route. The connector is mounted into the application that owns the middleware:

def role_from_session(request: Request) -> str:
    """Read the role stored in the session.

    Args:
        request: Incoming request.

    Returns:
        Stored role, ``guest`` before login.
    """
    role = request.session.get("userRole")
    return role if isinstance(role, str) else "guest"


def build_app() -> FastAPI:
    """Build the application.

    Returns:
        Application with session routes and the connector at ``/``.
    """
    app = FastAPI(openapi_url=None, docs_url=None, redoc_url=None)
    app.add_middleware(
        SessionMiddleware, secret_key=SECRET, max_age=SESSION_MAX_AGE
    )

    @app.get("/login/{role}")
    def login(
        request: Request, role: Literal["guest", "editor", "admin"]
    ) -> dict[str, str]:
        request.session["userRole"] = role
        return {"role": role}

    @app.get("/logout")
    def logout(request: Request) -> dict[str, str]:
        request.session.clear()
        return {"role": "guest"}

    @app.get("/whoami")
    def whoami(request: Request) -> dict[str, str]:
        return {"role": role_from_session(request)}

    # Own routes first: the connector serves every other /{action}.
    app.include_router(
        create_router(CONFIG, check_authentication=role_from_session)
    )
    return app

curl -c cookies.txt http://localhost:8081/login/editor
curl -b cookies.txt "http://localhost:8081/?action=permissions&source=uploads"
curl -b cookies.txt http://localhost:8081/logout

Full program: examples/session_auth.py.

Database lookup

from http import HTTPStatus

from starlette.requests import Request

from jcpy import HttpError


async def check_authentication(request: Request) -> str:
    api_key = request.headers.get("x-api-key")
    if api_key is None:
        return "guest"
    row = await db.fetchrow(
        "SELECT role FROM api_keys WHERE key = $1 AND active", api_key
    )
    if row is None:
        raise HttpError(HTTPStatus.UNAUTHORIZED, "Invalid API key")
    return row["role"]

The callback runs on every request; cache lookups that are expensive.

Security practices

1. Use HTTPS so tokens and cookies cannot be read on the way.
2. Verify, do not decode. Check signatures and expiry of tokens; never trust a role the client sent in plain form.
3. Keep secrets out of code: read signing keys from the environment or a secret store.
4. Deny by default. Give defaultRole (the role of anonymous requests) as little as possible, and allow more per role; see Access Control.
5. Rate-limit login and upload endpoints at the proxy or with middleware.
6. Consider onlyPOST when cookies authenticate requests: it keeps other sites from triggering actions with plain links and images.
7. Restrict CORS with allowedOrigins when allowCrossOrigin is on, especially together with cookies (credentials are allowed for accepted origins).