linux-cloud-router/debian-package/src/usr/lib/vpn-router/vpnrouter_passkeys.py

"""The passkeys of the router's users, for signing in to the Router Manager and the OpenID Provider (vpnrouter_oidc),
and for opening a shell, with python3-fido2.

A passkey is a discoverable WebAuthn credential with user verification, of the relying party
wan.local_fqdn, so the browser offers it at the HTTPS addresses on that name.
Its user handle is the user name. Each passkey of a user is a value of vpnRouterPasskey on the
user entry: a JSON object {"id", "credential", "label", "created"}, id the credential ID and
credential the attested credential data (with the public key), both base64url. A user has at
most PASSKEYS_MAX passkeys; service accounts have none.

Registration, sign-in and the proof of a signed-in user for a shell are two steps each: begin
answers the options for the browser and a pending id, whose state the web service keeps in memory
for PENDING_SECONDS; complete takes the id and the browser's answer. A request that cannot be done
raises vpnrouter_users.Failure, with the statuses of that module; a sign-in or a proof that fails
gives status 5 whatever the reason.
"""

import json
import secrets
import threading
import time

import fido2.features
import ldap
from fido2.server import Fido2Server
from fido2.utils import websafe_decode, websafe_encode
from fido2.webauthn import (AttestedCredentialData, PublicKeyCredentialRpEntity, PublicKeyCredentialUserEntity,
                            ResidentKeyRequirement, UserVerificationRequirement)

import vpnrouter
import vpnrouter_users

# The options and the browser's answers are in the JSON mapping of WebAuthn Level 3; python3-fido2 2.0 always uses it.
_json_mapping = getattr(fido2.features, 'webauthn_json_mapping', None)
if _json_mapping is not None:
    _json_mapping.enabled = True

#: The most passkeys of one user.
PASSKEYS_MAX = 10
#: The longest label of a passkey, in characters.
LABEL_MAX = 64
#: Seconds a begun registration or sign-in may be completed.
PENDING_SECONDS = 120
#: The most begun registrations and sign-ins kept at once.
PENDING_MAX = 1000

#: The begun registrations and sign-ins: pending id -> (kind, user name or None, state, monotonic time of expiry).
_pending = {}
_lock = threading.Lock()


def server():
    """The Fido2Server of the relying party wan.local_fqdn; passkeys need the name."""
    fqdn = vpnrouter.load_config()['wan']['local_fqdn'].strip()
    if not fqdn:
        raise vpnrouter_users.Failure('passkeys need wan.local_fqdn, the name the Router Manager is reached at', 2)
    return Fido2Server(rp=PublicKeyCredentialRpEntity(name=fqdn, id=fqdn))


def keep(kind, name, state):
    """Keep the state of a begun step and answer its pending id."""
    pending = secrets.token_urlsafe(24)
    now = time.monotonic()
    with _lock:
        for old in [key for key, (*_, expires) in _pending.items() if expires <= now]:
            del _pending[old]
        if len(_pending) >= PENDING_MAX:
            raise vpnrouter_users.Failure('too many passkey requests at once; retry in a moment')
        _pending[pending] = (kind, name, state, now + PENDING_SECONDS)
    return pending


def take(kind, name, pending):
    """The state of a begun step of the kind and user, used up; None when it is unknown or expired."""
    with _lock:
        found = _pending.pop(pending, None) if isinstance(pending, str) else None
    if found is None or found[0] != kind or found[1] != name or found[3] <= time.monotonic():
        return None
    return found[2]


def credentials(attributes):
    """The attested credential data of the passkeys of an entry."""
    return [AttestedCredentialData(websafe_decode(passkey['credential']))
            for _, passkey in vpnrouter_users.passkeys_of(attributes)]


def register_begin(connection, name):
    """Begin the registration of a passkey of the user: {"id", "options"}, options those of navigator.credentials.create."""
    current = vpnrouter_users.entry(connection, name)
    if vpnrouter_users.service(current):
        raise vpnrouter_users.Failure(f'{name} is a service account, which has no passkeys', 2)
    if len(vpnrouter_users.passkeys_of(current)) >= PASSKEYS_MAX:
        raise vpnrouter_users.Failure(f'user {name} has {PASSKEYS_MAX} passkeys, the most; remove one first', 2)
    options, state = server().register_begin(
        PublicKeyCredentialUserEntity(id=name.encode(), name=name, display_name=name),
        credentials(current),
        resident_key_requirement=ResidentKeyRequirement.REQUIRED,
        user_verification=UserVerificationRequirement.REQUIRED,
    )
    return {'id': keep('register', name, state), 'options': dict(options)}


def register_complete(connection, name, pending, response, label):
    """Store the passkey the browser created for a begun registration, with its label; the user (vpnrouter_users.summary)."""
    state = take('register', name, pending)
    if state is None:
        raise vpnrouter_users.Failure('the registration is unknown or expired; begin it again', 2)
    if label is None or (isinstance(label, str) and not label.strip()):
        label = 'Passkey'
    if not isinstance(label, str) or len(label.strip()) > LABEL_MAX:
        raise vpnrouter_users.Failure(f'the label must be text of at most {LABEL_MAX} characters', 2)
    if not isinstance(response, dict):
        raise vpnrouter_users.Failure('credential must be the answer of navigator.credentials.create as JSON', 2)
    try:
        data = server().register_complete(state, response).credential_data
    except (ValueError, KeyError, TypeError) as exc:
        raise vpnrouter_users.Failure(f'the passkey is refused: {exc}', 2)
    current = vpnrouter_users.entry(connection, name)
    value = json.dumps({
        'id': websafe_encode(data.credential_id),
        'credential': websafe_encode(bytes(data)),
        'label': label.strip(),
        'created': time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime()),
    }, separators=(',', ':'))
    changes = [] if 'vpnrouteraccount' in vpnrouter_users.classes(current) else [(ldap.MOD_ADD, 'objectClass', [b'vpnRouterAccount'])]
    changes.append((ldap.MOD_ADD, 'vpnRouterPasskey', [value.encode()]))
    connection.modify_s(vpnrouter_users.dn_of(name), changes)
    return vpnrouter_users.summary(vpnrouter_users.entry(connection, name))


def remove(connection, name, passkey_id=None):
    """Remove the user's passkey of the id, or all its passkeys without one; the user (vpnrouter_users.summary)."""
    current = vpnrouter_users.entry(connection, name)
    values = current.get('vpnRouterPasskey', [])
    kept = [value for value, passkey in vpnrouter_users.passkeys_of(current)
            if passkey_id is not None and passkey['id'] != passkey_id]
    if passkey_id is not None and len(kept) == len(vpnrouter_users.passkeys_of(current)):
        raise vpnrouter_users.Failure(f'user {name} has no passkey {passkey_id}', 3)
    if len(kept) != len(values):
        connection.modify_s(vpnrouter_users.dn_of(name), [(ldap.MOD_REPLACE, 'vpnRouterPasskey', kept)])
    return vpnrouter_users.summary(vpnrouter_users.entry(connection, name))


def login_begin():
    """Begin a sign-in with a passkey: {"id", "options"}, options those of navigator.credentials.get, for any user."""
    options, state = server().authenticate_begin(user_verification=UserVerificationRequirement.REQUIRED)
    return {'id': keep('login', None, state), 'options': dict(options)}


def login_complete(connection, pending, response):
    """The name of the user whose passkey signed a begun sign-in, when the user is enabled, not locked and not a
    service account."""
    refused = vpnrouter_users.Failure('the passkey is not accepted', 5)
    state = take('login', None, pending)
    if state is None or not isinstance(response, dict):
        raise refused
    try:
        name = websafe_decode(response['response']['userHandle']).decode()
    except (KeyError, TypeError, ValueError):
        raise refused
    if not vpnrouter_users.NAME.match(name):
        raise refused
    try:
        current = vpnrouter_users.entry(connection, name)
    except vpnrouter_users.Failure:
        raise refused
    user = vpnrouter_users.summary(current)
    if user['service'] or not user['enabled'] or user['locked']:
        raise refused
    try:
        server().authenticate_complete(state, credentials(current), response)
    except (ValueError, KeyError, TypeError):
        raise refused
    return name


def shell_begin(connection, name):
    """Begin the proof of the signed-in user for a shell: {"id", "options"}, options those of navigator.credentials.get
    for the user's own passkeys, with user verification."""
    own = credentials(vpnrouter_users.entry(connection, name))
    if not own:
        raise vpnrouter_users.Failure(f'{name} has no passkey; add one on the Account page', 2)
    options, state = server().authenticate_begin(own, user_verification=UserVerificationRequirement.REQUIRED)
    return {'id': keep('shell', name, state), 'options': dict(options)}


def shell_complete(connection, name, pending, response):
    """Check the browser's answer to a begun proof of the user for a shell: one of the user's own passkeys, with the
    user verified, while the user is enabled and not locked."""
    refused = vpnrouter_users.Failure('the passkey is not accepted', 5)
    state = take('shell', name, pending)
    if state is None or not isinstance(response, dict):
        raise refused
    current = vpnrouter_users.entry(connection, name)
    user = vpnrouter_users.summary(current)
    if not user['enabled'] or user['locked']:
        raise refused
    try:
        server().authenticate_complete(state, credentials(current), response)
    except (ValueError, KeyError, TypeError):
        raise refused