Skip to content

woodglue.apps.auth_pages

Browser login, logout, and the Identity Provider round trip.

Browser login and logout: /auth/login, /auth/logout, and the Identity Provider round trip at /auth/start/<provider> and /auth/callback/<provider> (see woodglue.apps.identity_providers).

Every way in ends in a Session and the wgl_session cookie (see woodglue.apps.authn). The login page offers a button per configured Identity Provider, and always a form for a Personal Access Token, e.g. the output of wgl token. A narrowed token is refused, since the Session it would start would carry all of the User's Grants.

/auth/cli and /auth/cli/token let wgl login get a Personal Access Token for the signed-in User: the browser approves on the server's page, which hands a one-time code to the CLI's loopback address; the CLI exchanges the code, with the PKCE verifier only it knows, for the token (see woodglue.cli_login). The token itself never appears in a URL.

The login pages accept a next path to return to; anything that is not a path on this server is replaced by /, so the login page cannot be used as an open redirect.

LoginHandler

Bases: RequestHandler

GET shows the login page; POST exchanges a token for a Session.

Source code in src/woodglue/apps/auth_pages.py
class LoginHandler(tornado.web.RequestHandler):
    """GET shows the login page; POST exchanges a token for a Session."""

    @override
    def get(self) -> None:
        self.write(_login_page(self, safe_next(self.get_argument("next", None))))

    @override
    def post(self) -> None:
        if not is_same_origin(self.request):
            raise tornado.web.HTTPError(403, "Cross-site login rejected")
        next_url = safe_next(self.get_argument("next", None))
        store: AuthStore = self.application.settings["auth_store"]
        caller = store.resolve(self.get_argument("token", "").strip())
        error = None
        if caller is None or caller.user_id is None or caller.credential_kind != "token":
            error = "That token is not valid."
        elif not caller.may_issue_credentials:
            error = "That token is narrowed; sign in with an unnarrowed token."
        if error is not None:
            self.set_status(401)
            self.write(_login_page(self, next_url, error))
            return
        assert caller is not None and caller.user_id is not None
        set_session_cookie(self, store.create_session(caller.user_id))
        self.redirect(next_url)

LogoutHandler

Bases: RequestHandler

POST ends the Session, closes its UI websockets, and returns to the login page.

Source code in src/woodglue/apps/auth_pages.py
class LogoutHandler(tornado.web.RequestHandler):
    """POST ends the Session, closes its UI websockets, and returns to the login page."""

    @override
    def post(self) -> None:
        if not is_same_origin(self.request):
            raise tornado.web.HTTPError(403, "Cross-site logout rejected")
        auth = authenticate(self)
        store: AuthStore = self.application.settings["auth_store"]
        if auth.session_secret is not None:
            store.end_session(auth.session_secret)
            sockets: UiSockets = self.application.settings["ui_sockets"]
            sockets.close_session(auth.session_secret)
        clear_session_cookie(self)
        self.redirect("/auth/login")

StartSignInHandler

Bases: RequestHandler

GET /auth/start/<provider>: send the browser to the Identity Provider.

Source code in src/woodglue/apps/auth_pages.py
class StartSignInHandler(tornado.web.RequestHandler):
    """GET `/auth/start/<provider>`: send the browser to the Identity Provider."""

    @override
    async def get(self, provider_id: str) -> None:
        client = _client(self, provider_id)
        pending = PendingSignIn(
            provider=provider_id,
            next_url=safe_next(self.get_argument("next", None)),
            redirect_uri=f"{_public_base(self)}/auth/callback/{provider_id}",
        )
        pending_sign_ins: PendingSignIns = self.application.settings["pending_sign_ins"]
        state = pending_sign_ins.add(pending)
        try:
            url = await client.authorize_url(pending, state)
        except SignInError as exc:
            logger.warning("Sign-in with %s could not start: %s", provider_id, exc)
            self.set_status(502)
            self.write(
                _login_page(self, pending.next_url, f"{client.config.display_name} is unavailable.")
            )
            return
        # Ties the callback to this browser, so nobody can complete a sign-in they did not start.
        self.set_cookie(
            LOGIN_COOKIE,
            state,
            path="/auth/",
            httponly=True,
            samesite="Lax",
            secure=public_scheme(self) == "https",
            expires_days=None,
        )
        self.redirect(url)

SignInCallbackHandler

Bases: RequestHandler

GET /auth/callback/<provider>: finish a sign-in and start a Session.

Source code in src/woodglue/apps/auth_pages.py
class SignInCallbackHandler(tornado.web.RequestHandler):
    """GET `/auth/callback/<provider>`: finish a sign-in and start a Session."""

    @override
    async def get(self, provider_id: str) -> None:
        client = _client(self, provider_id)
        state = self.get_argument("state", "")
        if not state or self.get_cookie(LOGIN_COOKIE) != state:
            raise tornado.web.HTTPError(400, "Sign-in state does not match this browser")
        self.clear_cookie(LOGIN_COOKIE, path="/auth/")
        pending_sign_ins: PendingSignIns = self.application.settings["pending_sign_ins"]
        pending = pending_sign_ins.pop(state)
        if pending is None or pending.provider != provider_id:
            raise tornado.web.HTTPError(400, "Sign-in expired; start again")
        name = client.config.display_name
        if self.get_argument("error", None):
            self._fail(401, pending, f"{name} did not sign you in: {self.get_argument('error')}")
            return
        try:
            who = await client.complete(pending, self.get_argument("code", ""))
        except SignInError as exc:
            logger.warning("Sign-in with %s failed: %s", provider_id, exc)
            self._fail(401, pending, f"Sign-in with {name} failed.")
            return
        if not admits(client.config.admit, who):
            logger.info("Sign-in with %s refused for %s (%s)", provider_id, who.subject, who.email)
            self._fail(403, pending, f"Your {name} account is not allowed on this server.")
            return
        store: AuthStore = self.application.settings["auth_store"]
        auth_config: AuthConfig = self.application.settings["config"].auth
        try:
            user = store.sign_in_person(
                provider_id,
                who.subject,
                email=who.email,
                email_verified=who.email_verified,
                name=who.name,
                role_sources=role_sources(client.config, who, auth_config.admins),
            )
        except UserDisabled:
            self._fail(403, pending, "Your account is disabled.")
            return
        logger.info("Signed in %s (%s) with %s", user.id, user.email, provider_id)
        set_session_cookie(self, store.create_session(user.id))
        self.redirect(pending.next_url)

    def _fail(self, status: int, pending: PendingSignIn, error: str) -> None:
        self.set_status(status)
        self.write(_login_page(self, pending.next_url, error))

CliCodes

One-time codes handed to a CLI's loopback address, by code.

Source code in src/woodglue/apps/auth_pages.py
class CliCodes:
    """One-time codes handed to a CLI's loopback address, by code."""

    _codes: dict[str, _CliCode]

    def __init__(self) -> None:
        self._codes = {}

    def add(self, user_id: str, name: str, challenge: str) -> str:
        """Issue a code for an approved request, dropping expired ones."""
        cutoff = time.monotonic() - _CLI_CODE_TTL_SECONDS
        self._codes = {c: v for c, v in self._codes.items() if v.created >= cutoff}
        code = secrets.token_urlsafe(24)
        self._codes[code] = _CliCode(user_id, name, challenge)
        return code

    def redeem(self, code: str, verifier: str) -> _CliCode | None:
        """
        The approval behind `code` if `verifier` matches its challenge. Any
        attempt uses the code up, so a wrong verifier cannot be retried.
        """
        entry = self._codes.pop(code, None)
        if entry is None or time.monotonic() - entry.created > _CLI_CODE_TTL_SECONDS:
            return None
        if not secrets.compare_digest(pkce_challenge(verifier), entry.challenge):
            return None
        return entry

add(user_id, name, challenge)

Issue a code for an approved request, dropping expired ones.

Source code in src/woodglue/apps/auth_pages.py
def add(self, user_id: str, name: str, challenge: str) -> str:
    """Issue a code for an approved request, dropping expired ones."""
    cutoff = time.monotonic() - _CLI_CODE_TTL_SECONDS
    self._codes = {c: v for c, v in self._codes.items() if v.created >= cutoff}
    code = secrets.token_urlsafe(24)
    self._codes[code] = _CliCode(user_id, name, challenge)
    return code

redeem(code, verifier)

The approval behind code if verifier matches its challenge. Any attempt uses the code up, so a wrong verifier cannot be retried.

Source code in src/woodglue/apps/auth_pages.py
def redeem(self, code: str, verifier: str) -> _CliCode | None:
    """
    The approval behind `code` if `verifier` matches its challenge. Any
    attempt uses the code up, so a wrong verifier cannot be retried.
    """
    entry = self._codes.pop(code, None)
    if entry is None or time.monotonic() - entry.created > _CLI_CODE_TTL_SECONDS:
        return None
    if not secrets.compare_digest(pkce_challenge(verifier), entry.challenge):
        return None
    return entry

CliAuthorizeHandler

Bases: RequestHandler

GET shows the approval page; POST issues a one-time code to the CLI's loopback address.

Source code in src/woodglue/apps/auth_pages.py
class CliAuthorizeHandler(tornado.web.RequestHandler):
    """GET shows the approval page; POST issues a one-time code to the CLI's loopback address."""

    def _params(self) -> dict[str, str]:
        params = {
            k: self.get_argument(k, "") for k in ("redirect_uri", "state", "code_challenge", "name")
        }
        if not is_loopback_redirect(params["redirect_uri"]) or not params["code_challenge"]:
            raise tornado.web.HTTPError(400, "redirect_uri must be a loopback address with a port")
        params["name"] = params["name"].strip() or "wgl"
        return params

    def _caller(self) -> Caller | None:
        caller = authenticate(self).caller
        if caller is None:
            self.redirect(f"/auth/login?next={quote(self.request.uri or '/', safe='')}")
        elif caller.user_id is None:
            raise tornado.web.HTTPError(400, "Auth is disabled; the CLI needs no token")
        elif not caller.may_issue_credentials:
            raise tornado.web.HTTPError(403, "This credential cannot create tokens")
        return caller

    @override
    def get(self) -> None:
        params = self._params()
        caller = self._caller()
        if caller is None:
            return
        fields = "".join(
            f'<input type="hidden" name="{k}" value="{html.escape(v, quote=True)}">'
            for k, v in params.items()
        )
        self.write(
            _CONFIRM_PAGE.format(
                name=html.escape(params["name"]),
                user=html.escape(caller.email or caller.name),
                fields=fields,
            )
        )

    @override
    def post(self) -> None:
        if not is_same_origin(self.request):
            raise tornado.web.HTTPError(403, "Cross-site authorization rejected")
        params = self._params()
        caller = self._caller()
        if caller is None:
            return
        assert caller.user_id is not None
        codes: CliCodes = self.application.settings["cli_codes"]
        code = codes.add(caller.user_id, params["name"], params["code_challenge"])
        self.redirect(
            f"{params['redirect_uri']}?{urlencode({'code': code, 'state': params['state']})}"
        )

CliTokenHandler

Bases: RequestHandler

POST {code, code_verifier}: the Personal Access Token for an approved code.

Source code in src/woodglue/apps/auth_pages.py
class CliTokenHandler(tornado.web.RequestHandler):
    """POST `{code, code_verifier}`: the Personal Access Token for an approved code."""

    @override
    def post(self) -> None:
        try:
            body = json.loads(self.request.body)
            code, verifier = str(body["code"]), str(body["code_verifier"])
        except (ValueError, KeyError, TypeError):
            raise tornado.web.HTTPError(400, "Expected JSON with code and code_verifier") from None
        codes: CliCodes = self.application.settings["cli_codes"]
        entry = codes.redeem(code, verifier)
        if entry is None:
            raise tornado.web.HTTPError(400, "Unknown, expired, or already used code")
        store: AuthStore = self.application.settings["auth_store"]
        issued = store.create_token(entry.user_id, entry.name)
        logger.info("Issued CLI token %s to %s", issued.id, entry.user_id)
        self.write({"token": issued.secret, "id": issued.id})

safe_next(target)

A local path to redirect to after login, or /.

>>> safe_next("/console/#runs"), safe_next("//evil.example"), safe_next("https://x/")
('/console/#runs', '/', '/')
>>> safe_next("/\\evil.example"), safe_next(None)
('/', '/')
Source code in src/woodglue/apps/auth_pages.py
def safe_next(target: str | None) -> str:
    """
    A local path to redirect to after login, or `/`.

    >>> safe_next("/console/#runs"), safe_next("//evil.example"), safe_next("https://x/")
    ('/console/#runs', '/', '/')
    >>> safe_next("/\\\\evil.example"), safe_next(None)
    ('/', '/')
    """
    if not target or not target.startswith("/") or target[1:2] in ("/", "\\"):
        return "/"
    return target

render_login_page(next_url, error=None, providers_html='')

The login page HTML: provider buttons (if any) above the token form.

Source code in src/woodglue/apps/auth_pages.py
def render_login_page(next_url: str, error: str | None = None, providers_html: str = "") -> str:
    """The login page HTML: provider buttons (if any) above the token form."""
    intro = (
        "Choose how to sign in."
        if providers_html
        else "Paste a Personal Access Token, e.g. the output of <code>wgl token</code>."
    )
    return _PAGE.format(
        intro=intro,
        error=f'<p class="error">{html.escape(error)}</p>' if error else "",
        providers=providers_html,
        next=html.escape(next_url, quote=True),
    )

is_loopback_redirect(uri)

Whether uri points at a CLI listening on this computer.

>>> is_loopback_redirect("http://127.0.0.1:53111/cb"), is_loopback_redirect("http://localhost:9/")
(True, True)
>>> is_loopback_redirect("http://127.0.0.1/cb"), is_loopback_redirect("https://evil.example:9/")
(False, False)
Source code in src/woodglue/apps/auth_pages.py
def is_loopback_redirect(uri: str) -> bool:
    """
    Whether `uri` points at a CLI listening on this computer.

    >>> is_loopback_redirect("http://127.0.0.1:53111/cb"), is_loopback_redirect("http://localhost:9/")
    (True, True)
    >>> is_loopback_redirect("http://127.0.0.1/cb"), is_loopback_redirect("https://evil.example:9/")
    (False, False)
    """
    parts = urlsplit(uri)
    return (
        parts.scheme == "http"
        and parts.hostname in ("127.0.0.1", "localhost", "::1")
        and parts.port is not None
    )