CLI

authkeys.cli

Command line, built on duho.

authkeys [-c FILE] resolve [user] prints a user's authorized keys (the AuthorizedKeysCommand entry point); resolve is the default command, so authkeys %u works in sshd_config. check, serve, cache and completion are the other commands.

__all__ = ['AuthKeysCLI', 'Cache', 'Check', 'Resolve', 'Serve', 'main'] module-attribute

AuthKeysCLI

Bases: ConfigArgs, LoggingArgs, Cli

Pluggable AuthorizedKeysCommand provider for OpenSSH.

Cache

Bases: LoggingArgs, Cmd

Inspect, purge, or warm the resolved-keys cache.

action = 'show' class-attribute instance-attribute

One of: show, purge, warm

all = False class-attribute instance-attribute

purge: remove every entry

dry_run = False class-attribute instance-attribute

purge: list what would be removed, remove nothing

expired = False class-attribute instance-attribute

purge: only remove expired entries

source = None class-attribute instance-attribute

Limit purge to this source

user = None class-attribute instance-attribute

Limit purge to this user (uid), or warm this user

users = [] class-attribute instance-attribute

warm: users to pre-resolve into the cache

__call__()

Source code in src/authkeys/cli/cache.py

def __call__(self) -> "int | None":
    auth = build(self)
    if auth.cache is None:
        raise CommandError("No cache configured", 3)
    backend = auth.cache.backend
    try:
        if self.action == "show":
            for uid, src in sorted(backend.keys()):
                entry = backend[(uid, src)]
                age = int(time.time() - entry[1]) if entry else "?"
                print(f"{src}\t{uid}\tage={age}s")
            return 0
        if self.action == "purge":
            return self._purge(auth.cache)
        targets = self.users or ([self.user] if self.user else [])
        for username in targets:
            keys = auth.resolve(username)
            self._logger_.info(f"Warmed {username}: {len(keys)} key(s)")
        return 0
    except NotImplementedError:
        raise CommandError(
            "The configured cache backend does not support this action "
            "(in-memory caches are per-process).",
            3,
        ) from None

Check

Bases: Resolve

Resolve a user's keys with per-source tracing on stderr (debugging).

__call__()

Source code in src/authkeys/cli/resolve.py

def __call__(self) -> "int | None":
    # The engine logs which source produced which key at INFO; raise this
    # command's logger so that tracing reaches stderr.
    self._logger_.setLevel("DEBUG")
    return super().__call__()

Resolve

Bases: LoggingArgs, Cmd

Print the authorized SSH keys for a user (AuthorizedKeysCommand).

format = 'authorized_keys' class-attribute instance-attribute

Output format: authorized_keys (default, sshd wire format) or json

username = None class-attribute instance-attribute

User to resolve (defaults to the current user)

__call__()

Source code in src/authkeys/cli/resolve.py

def __call__(self) -> "int | None":
    # An AuthorizedKeysCommand must never dump a traceback into auth.log: any
    # error is one line on stderr and status 3; sshd reads a non-zero status
    # or empty output as "no keys". Status 0 includes "no keys found".
    try:
        auth = build(self)
        username = self.username or utils.get_user().pw_name
        print_keys(list(auth.resolve(username)), self.format)
        return 0
    except SystemExit:
        raise
    except BaseException as e:  # noqa: BLE001 - deliberate catch-all boundary
        self._logger_.error(f"authkeys {self._verb_} failed: {e}")
        return 3

Serve

Bases: LoggingArgs, Cmd

Run the unattended HTTP key server.

Every option may also be set in the [serve] table of the configuration or through AUTHKEYS_SERVE_: the command line wins, then the variable, then the file. The API key comes only from [serve] api_key (use ${env.NAME}), never from the command line, where other users can read it.

certificate = None class-attribute instance-attribute

PEM certificate (chain) to serve HTTPS with

key = None class-attribute instance-attribute

PEM private key for --certificate

listen = '127.0.0.1:8090' class-attribute instance-attribute

Where to listen: host:port, [v6]:port or :port, comma-separated (port 8090 by default)

max_connections = 64 class-attribute instance-attribute

Most requests served at once (more get 503)

max_usernames = 16 class-attribute instance-attribute

Most username parameters one request may carry

path = '/keys' class-attribute instance-attribute

URL path that serves keys

request_timeout = 10.0 class-attribute instance-attribute

Seconds a connection may stay idle

__call__()

Source code in src/authkeys/cli/serve.py

def __call__(self) -> "int | None":
    from ..server import bind_servers, serve

    settings = settings_for(self)
    auth = AuthKeys()
    auth.load_config(settings)
    # Fail closed: an api_key line that resolved empty (an unset ${env.NAME},
    # a typo) must not start the server with authentication disabled.
    has_key_line = "api_key" in settings.serve
    api_key = settings.serve.get("api_key") or None
    if has_key_line and not api_key:
        raise CommandError(
            "authkeys serve: 'api_key' is configured in [serve] but resolved to "
            "an empty value (unset ${env.NAME}?). Refusing to start with "
            "authentication silently disabled. Set the key, or remove the "
            "'api_key' line to run without authentication.",
            1,
        )
    context = None
    if self.key and not self.certificate:
        raise CommandError("authkeys serve: --key needs --certificate", 2)
    try:
        if self.certificate:
            import sslcontext

            context = sslcontext.create_context(
                "server", certificate=self.certificate, key=self.key
            )
        servers = bind_servers(
            auth,
            self.listen,
            api_key=api_key,
            path=self.path,
            max_usernames=self.max_usernames,
            require_auth=has_key_line,
            request_timeout=self.request_timeout,
            max_connections=self.max_connections,
            ssl_context=context,
        )
    except (ValueError, OSError) as e:
        raise CommandError(f"authkeys serve: {e}", 3) from None
    serve(servers)
    return 0

main(argv=None)

Run the CLI with argv (default sys.argv[1:]) and return the status.

Source code in src/authkeys/cli/__init__.py

def main(argv: "_ty.Optional[_ty.Sequence[str]]" = None) -> int:
    """Run the CLI with ``argv`` (default ``sys.argv[1:]``) and return the status."""
    return _duho_run(
        AuthKeysCLI, list(sys.argv[1:] if argv is None else argv), exit=False
    )