Sources

authkeys.sources

Key sources, found by name.

backend = "<name>" in a [source.<name>] table selects one. file, http, github and ldap are built in; an installed distribution adds a source through the authkeys.sources entry-point group (name = "pkg.module:Class", a :class:~authkeys.KeySource subclass). Optional dependencies (syshttp, ldap3, cryptography) are imported inside each source, so naming a source never fails for a missing extra until it is used.

__all__ = ['FileSource', 'GitHubSource', 'HTTPSource', 'LDAPSource', 'get_source', 'only_auth_keys', 'register_source', 'source_names'] module-attribute

FileSource(options, globals)

Bases: KeySource

Keys from the target user's own ~/.ssh files.

Source code in src/authkeys/sources/_file.py

def __init__(self, options: Mapping[str, Any], globals: dict) -> None:
    paths = typed(options, "paths", (list, str), _DEFAULT_PATHS, "file source")
    self.paths = [paths] if isinstance(paths, str) else [str(p) for p in paths]

options = frozenset({'paths'}) class-attribute instance-attribute

paths = [paths] if isinstance(paths, str) else [str(p) for p in paths] instance-attribute

authorized_keys(username)

Source code in src/authkeys/sources/_file.py

def authorized_keys(self, username: str) -> Iterable[str]:
    import pwd

    try:
        user = pwd.getpwnam(username)
    except KeyError:
        return
    home = Path(user.pw_dir)
    ssh_home = home / ".ssh"
    for path in self.paths:
        text = utils.read_user_file(ssh_home / path, user.pw_uid, home)
        for line in (text or "").splitlines():
            line = line.strip()
            if line and not line.startswith("#"):
                yield line

GitHubSource(options, globals, *, ssl_context=None)

Bases: KeySource

Keys from a forge's plain-text .keys endpoint.

Source code in src/authkeys/sources/_github.py

def __init__(
    self,
    options: Mapping[str, Any],
    globals: dict,
    *,
    ssl_context: "Optional[ssl.SSLContext]" = None,
) -> None:
    self.url = typed(options, "url", str, _DEFAULT_URL, "github source")
    self.client = KeyClient(options, "github source", ssl_context=ssl_context)

client = KeyClient(options, 'github source', ssl_context=ssl_context) instance-attribute

options = frozenset({'url'}) | CLIENT_OPTIONS class-attribute instance-attribute

url = typed(options, 'url', str, _DEFAULT_URL, 'github source') instance-attribute

authorized_keys(username)

Source code in src/authkeys/sources/_github.py

def authorized_keys(self, username: str) -> Iterable[str]:
    # Percent-encode the username: it may come from a `serve` request and
    # must not alter the URL path or query.
    quoted = urllib.parse.quote(username, safe="")
    return self.client.lines(self.url.format(username=quoted))

HTTPSource(options, globals, *, ssl_context=None)

Bases: KeySource

Keys from an HTTP endpoint, one authorized_keys line per row.

ssl_context (programmatic use) replaces the TLS options.

Source code in src/authkeys/sources/_http.py

def __init__(
    self,
    options: Mapping[str, Any],
    globals: dict,
    *,
    ssl_context: "Optional[ssl.SSLContext]" = None,
) -> None:
    self.url = typed(options, "url", str, None, "http source")
    if not self.url:
        raise ConfigError("the http source needs a 'url'")
    self.client = KeyClient(options, "http source", ssl_context=ssl_context)

client = KeyClient(options, 'http source', ssl_context=ssl_context) instance-attribute

options = frozenset({'url'}) | CLIENT_OPTIONS class-attribute instance-attribute

url = typed(options, 'url', str, None, 'http source') instance-attribute

authorized_keys(username)

Source code in src/authkeys/sources/_http.py

def authorized_keys(self, username: str) -> Iterable[str]:
    # Percent-encode the username: it may come from a `serve` request and
    # must not alter the URL path or query.
    quoted = urllib.parse.quote(username, safe="")
    return self.client.lines(self.url.format(username=quoted))

LDAPSource(options, globals, *, ssl_context=None)

Bases: KeySource

Keys derived from X.509 certificates stored in an LDAP directory.

ssl_context (programmatic use) is the client TLS context; it selects the ldap3 client, which can take one.

Source code in src/authkeys/sources/_ldap.py

def __init__(
    self,
    options: Mapping[str, Any],
    globals: dict,
    *,
    ssl_context: "Optional[ssl.SSLContext]" = None,
) -> None:
    where = "ldap source"
    server = typed(options, "server", str, None, where)
    if not server:
        raise ConfigError("the ldap source needs a 'server'")
    auth = typed(options, "auth", str, "gssapi", where).strip().lower()
    if auth not in _AUTH_MODES:
        raise ConfigError(
            f"ldap source: auth must be one of {', '.join(_AUTH_MODES)}, not {auth!r}"
        )
    starttls = typed(options, "starttls", bool, False, where)
    url = server.strip()
    if "://" not in url:
        url = f"ldaps://{url}"
    scheme = url.split("://", 1)[0].lower()
    if scheme not in ("ldap", "ldaps"):
        raise ConfigError(
            f"ldap source: 'server' must be ldaps:// or ldap://, not {scheme}://"
        )
    # A clear-text connection decides who logs in: allow it only when Kerberos
    # seals the session or StartTLS encrypts it.
    if scheme == "ldap" and not (starttls or auth == "gssapi"):
        raise ConfigError(
            "ldap source 'server' must be an ldaps:// URL or a host name, or use "
            'ldap:// with starttls = true or auth = "gssapi" (a sealed session)'
        )
    tls_cert = typed(options, "tls_cert", str, None, where)
    tls_key = typed(options, "tls_key", str, None, where)
    if auth == "external" and not (tls_cert or ssl_context):
        raise ConfigError(
            'ldap source: auth = "external" needs tls_cert and tls_key'
        )
    bind_dn = typed(options, "bind_dn", str, None, where)
    password_env = typed(options, "password_env", str, None, where)
    if auth == "simple":
        if not (bind_dn and password_env):
            raise ConfigError(
                'ldap source: auth = "simple" needs bind_dn and password_env'
            )
        if scheme == "ldap" and not starttls:
            raise ConfigError(
                "ldap source: a simple bind needs TLS (ldaps:// or starttls)"
            )
    keytab = typed(options, "keytab", str, _DEFAULT_KEYTAB, where)
    if auth == "gssapi":
        use_machine_keytab(keytab)
    self.username_attr = typed(options, "username_attr", str, "uid", where)
    self.cert_attr = typed(options, "cert_attr", str, "userCertificate", where)
    self.settings = LDAPSettings(
        url=url,
        starttls=starttls,
        basedn=typed(options, "basedn", str, "", where),
        attribute=self.cert_attr,
        auth=auth,
        keytab=keytab,
        principal=typed(options, "principal", str, None, where),
        bind_dn=bind_dn,
        password_env=password_env,
        tls_verify=typed(options, "tls_verify", str, None, where),
        tls_cert=tls_cert,
        tls_key=tls_key,
        timeout=float(typed(options, "timeout", (int, float), 10.0, where)),
        ssl_context=ssl_context,
    )
    self.client = make_client(
        self.settings, typed(options, "client", str, "auto", where)
    )
    cert_filter: "Optional[Union[str, CertFilter]]" = typed(
        options, "cert_filter", str, None, where
    )
    if cert_filter is None:
        cert_filter = only_auth_keys
    elif cert_filter.strip().lower() == "none":
        cert_filter = None
    else:
        cert_filter = utils.import_module_object(cert_filter)
    self.cert_filter = cert_filter
    self.issuers = self._load_issuers(typed(options, "issuer_ca", str, None, where))

cert_attr = typed(options, 'cert_attr', str, 'userCertificate', where) instance-attribute

cert_filter = cert_filter instance-attribute

client = make_client(self.settings, typed(options, 'client', str, 'auto', where)) instance-attribute

issuers = self._load_issuers(typed(options, 'issuer_ca', str, None, where)) instance-attribute

options = frozenset({'server', 'basedn', 'cert_attr', 'username_attr', 'client', 'auth', 'keytab', 'principal', 'bind_dn', 'password_env', 'starttls', 'tls_verify', 'tls_cert', 'tls_key', 'timeout', 'cert_filter', 'issuer_ca'}) class-attribute instance-attribute

settings = LDAPSettings(url=url, starttls=starttls, basedn=typed(options, 'basedn', str, '', where), attribute=self.cert_attr, auth=auth, keytab=keytab, principal=typed(options, 'principal', str, None, where), bind_dn=bind_dn, password_env=password_env, tls_verify=typed(options, 'tls_verify', str, None, where), tls_cert=tls_cert, tls_key=tls_key, timeout=float(typed(options, 'timeout', (int, float), 10.0, where)), ssl_context=ssl_context) instance-attribute

username_attr = typed(options, 'username_attr', str, 'uid', where) instance-attribute

authorized_keys(username)

Source code in src/authkeys/sources/_ldap.py

def authorized_keys(self, username: str) -> Iterable[str]:
    from cryptography import x509
    from cryptography.hazmat.primitives.serialization.ssh import (
        serialize_ssh_public_key,
    )

    keys: List[str] = []
    for raw in self.client.certificates(self._search_filter(username)):
        cert = x509.load_der_x509_certificate(bytes(raw))
        if self.issuers and not self._issued_by_trusted(cert):
            continue
        comment: "Union[bool, str, None]" = None
        if self.cert_filter:
            comment = self.cert_filter(username, cert)
            if comment is False:
                continue
        if comment is None or isinstance(comment, bool):
            # The comment is one field of an authorized_keys line: no spaces.
            subject = cert.subject.rfc4514_string().replace(" ", "_")
            comment = f"{username}({subject})"
        key = serialize_ssh_public_key(cert.public_key()).decode()
        keys.append(f"{key} {comment}")
    return keys

get_source(name)

The source class registered as name; ConfigError listing the names.

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

def get_source(name: str) -> "Type[KeySource]":
    """The source class registered as ``name``; ``ConfigError`` listing the names."""
    _load_plugins()
    try:
        return _SOURCES[name]
    except KeyError:
        raise ConfigError(
            f"unknown source backend {name!r}; registered: {', '.join(source_names())}"
        ) from None

only_auth_keys(username, cert)

Default filter: keep only certificates that are login credentials.

A certificate is kept when it is within its validity period, is not self-signed, its KeyUsage (when present) permits digital signature, and its ExtendedKeyUsage (when present) includes clientAuth or anyExtendedKeyUsage. Directories also hold encryption, e-mail and recovery certificates in the same attribute; those must not become SSH keys.

Source code in src/authkeys/sources/_ldap.py

def only_auth_keys(username: str, cert) -> bool:
    """Default filter: keep only certificates that are login credentials.

    A certificate is kept when it is within its validity period, is not
    self-signed, its KeyUsage (when present) permits digital signature, and its
    ExtendedKeyUsage (when present) includes clientAuth or anyExtendedKeyUsage.
    Directories also hold encryption, e-mail and recovery certificates in the
    same attribute; those must not become SSH keys.
    """
    from cryptography import x509
    from cryptography.x509.oid import ExtendedKeyUsageOID

    before, after = _validity(cert)
    if not before <= datetime.now(timezone.utc) <= after:
        return False
    if cert.issuer == cert.subject:
        return False
    try:
        usage = cert.extensions.get_extension_for_class(x509.KeyUsage).value
    except x509.ExtensionNotFound:
        usage = None
    if usage is not None and not usage.digital_signature:
        return False
    try:
        purposes = cert.extensions.get_extension_for_class(x509.ExtendedKeyUsage).value
    except x509.ExtensionNotFound:
        purposes = None
    if purposes is not None and not (
        ExtendedKeyUsageOID.CLIENT_AUTH in purposes
        or ExtendedKeyUsageOID.ANY_EXTENDED_KEY_USAGE in purposes
    ):
        return False
    return True

register_source(name, cls)

Register a source class under name; a name taken by another class is a ValueError. Returns cls.

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

def register_source(name: str, cls: "Type[KeySource]") -> "Type[KeySource]":
    """Register a source class under ``name``; a name taken by another class is a
    ``ValueError``. Returns ``cls``."""
    if _SOURCES.get(name, cls) is not cls:
        raise ValueError(f"source {name!r} is already registered")
    _SOURCES[name] = cls
    return cls

source_names()

The registered source names, sorted.

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

def source_names() -> "Tuple[str, ...]":
    """The registered source names, sorted."""
    _load_plugins()
    return tuple(sorted(_SOURCES))