Skip to content

Sources

authkeys.sources.file

File source: read keys from a user's ~/.ssh/authorized_keys files.

AuthorizedKeysFiles(conf, globals)

Bases: AuthkeysSource

Source code in src/authkeys/sources/file.py 
11
12
def __init__(self, conf: SectionProxy, globals) -> None:
    self.paths = conf.getlist("paths", ["authorized_keys", "authorized_keys2"])

paths = conf.getlist('paths', ['authorized_keys', 'authorized_keys2']) instance-attribute

authorized_keys(username)

Source code in src/authkeys/sources/file.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
def authorized_keys(self, username: str) -> Iterable[str]:
    import pwd

    try:
        user = pwd.getpwnam(username)
    except KeyError:
        return
    ssh_home = Path(user.pw_dir) / ".ssh"
    for path in self.paths:
        keyfile = ssh_home / path
        if keyfile.exists():
            for line in keyfile.read_text().splitlines():
                line = line.strip()
                if line and not line.startswith("#"):
                    yield line

authkeys.sources.http

HTTP source: fetch keys from a URL, one authorized_keys line per row.

The address is a format string receiving {username}. Optional mutual-TLS client credentials are given as credentials = cert.pem,key.pem. Requires the optional requests dependency (pip install authkeys[http]).

HttpAuthorizedKeys(conf, globals)

Bases: AuthkeysSource

Source code in src/authkeys/sources/http.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
def __init__(self, conf: SectionProxy, globals) -> None:
    import requests

    self.address = conf.get("address")
    if not self.address:
        raise ValueError("http source requires an 'address'")
    self.session = requests.Session()
    self.timeout = conf.getfloat("timeout", 10.0)
    self.options: dict = {}
    credentials = conf.get("credentials", None)
    if credentials:
        self.options["cert"] = tuple(c.strip() for c in credentials.split(","))
    verify = conf.get("verify", None)
    if verify is not None:
        self.options["verify"] = _parse_verify(verify)

address = conf.get('address') instance-attribute

options = {} instance-attribute

session = requests.Session() instance-attribute

timeout = conf.getfloat('timeout', 10.0) instance-attribute

authorized_keys(username)

Source code in src/authkeys/sources/http.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def authorized_keys(self, username: str) -> Iterable[str]:
    # Percent-encode the username: it may be attacker-controlled (via
    # `authkeys serve`) and must not alter the URL path/query.
    quoted = urllib.parse.quote(username, safe="")
    resp = self.session.get(
        self.address.format(username=quoted),
        timeout=self.timeout,
        **self.options,
    )
    # Raise on any non-success status so a transient 5xx/403 routes through
    # the resolver's error path (and expired_on_error) instead of being
    # silently cached as "this user has no keys".
    resp.raise_for_status()
    for line in resp.content.decode().splitlines():
        line = line.strip()
        if line and not line.startswith("#"):
            yield line

authkeys.sources.github

GitHub .keys source: fetch a user's public keys from github.com.

GitHub (and GitLab, and similar forges) publish a user's public SSH keys at a plain-text .keys URL, one key per line, WITHOUT comments. The url template receives {username} and defaults to the GitHub endpoint; point it at https://gitlab.com/{username}.keys (or any other forge with the same convention) to reuse this source without a separate class.

GithubKeys(conf, globals)

Bases: AuthkeysSource

Source code in src/authkeys/sources/github.py 
20
21
22
23
24
25
def __init__(self, conf: SectionProxy, globals) -> None:
    import requests

    self.url = conf.get("url", _DEFAULT_URL)
    self.session = requests.Session()
    self.timeout = conf.getfloat("timeout", 10.0)

session = requests.Session() instance-attribute

timeout = conf.getfloat('timeout', 10.0) instance-attribute

url = conf.get('url', _DEFAULT_URL) instance-attribute

authorized_keys(username)

Source code in src/authkeys/sources/github.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
def authorized_keys(self, username: str) -> Iterable[str]:
    # Percent-encode the username: it may be attacker-controlled (via
    # `authkeys serve`) and must not alter the URL path/query.
    quoted = urllib.parse.quote(username, safe="")
    resp = self.session.get(
        self.url.format(username=quoted),
        timeout=self.timeout,
    )
    # Raise on any non-success status so a transient 5xx/403 routes through
    # the resolver's error path (and expired_on_error) instead of being
    # silently cached as "this user has no keys".
    resp.raise_for_status()
    for line in resp.content.decode().splitlines():
        line = line.strip()
        if line and not line.startswith("#"):
            yield line

authkeys.sources.ldap

LDAP source: derive SSH keys from X.509 certificates stored in a directory.

Each matching entry's cert_attr values are parsed as DER certificates; the public key is serialized to OpenSSH format. An optional cert_filter callable (username, cert) -> bool | str can drop a certificate or supply a comment.

Requires the optional ldap extra (pip install authkeys[ldap]: ldap3 + cryptography).

CertFilter = Callable[[str, 'object'], 'Union[bool, str]'] module-attribute

LOGGER = logging.getLogger('authkeys.sources.ldap') module-attribute

LdapAuthorizedKeys(conf, globals)

Bases: AuthkeysSource

Source code in src/authkeys/sources/ldap.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
def __init__(self, conf: SectionProxy, globals) -> None:
    # tls_cert/tls_key are the *client* mutual-TLS credentials (not server
    # verification). Server verification is controlled by tls_verify below.
    self.tls_cert = conf.get("tls_cert", "/etc/pki/tls/certs/server.crt")
    self.tls_key = conf.get("tls_key", "/etc/pki/tls/private/server.key")
    # tls_verify: 'none' (insecure opt-out), 'system'/unset (verify against the
    # OS trust store), or a path (file -> CA bundle, dir -> CA directory).
    # Default is to VERIFY -- an unverified LDAPS channel decides who logs in.
    self.tls_verify = conf.get("tls_verify", None)
    self.timeout = conf.getfloat("timeout", 10.0)
    self.server = conf.get("server", None)
    self.basedn = conf.get("basedn", "")
    self.cert_attr = conf.get("cert_attr", "userCertificate")
    self.username_attr = conf.get("username_attr", "uid")
    cert_filter: "Optional[Union[str, CertFilter]]" = conf.get("cert_filter", None)
    if isinstance(cert_filter, str):
        cert_filter = utils.import_module_object(cert_filter)
    self.cert_filter = cert_filter

basedn = conf.get('basedn', '') instance-attribute

cert_attr = conf.get('cert_attr', 'userCertificate') instance-attribute

cert_filter = cert_filter instance-attribute

server = conf.get('server', None) instance-attribute

timeout = conf.getfloat('timeout', 10.0) instance-attribute

tls_cert = conf.get('tls_cert', '/etc/pki/tls/certs/server.crt') instance-attribute

tls_key = conf.get('tls_key', '/etc/pki/tls/private/server.key') instance-attribute

tls_verify = conf.get('tls_verify', None) instance-attribute

username_attr = conf.get('username_attr', 'uid') instance-attribute

authorized_keys(username)

Source code in src/authkeys/sources/ldap.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
def authorized_keys(self, username: str) -> Iterable[str]:
    import ldap3
    from cryptography import x509
    from cryptography.hazmat.primitives.serialization.ssh import (
        serialize_ssh_public_key,
    )

    if (
        not self.server
        or not Path(self.tls_key).exists()
        or not Path(self.tls_cert).exists()
    ):
        raise RuntimeError("LDAP source is missing required configuration")

    tls = ldap3.Tls(self.tls_key, self.tls_cert, **self._tls_kwargs())
    server = ldap3.Server(
        self.server, use_ssl=True, tls=tls, connect_timeout=self.timeout
    )
    # receive_timeout bounds the search so a black-holed server can't hang
    # the resolve path (and, in `serve`, other requests behind it).
    conn = ldap3.Connection(
        server=server, auto_bind=True, receive_timeout=self.timeout
    )
    try:
        found = conn.search(
            self.basedn,
            self._search_filter(username),
            attributes=[self.cert_attr],
        )
        if not found or not conn.response:
            return
        for raw in conn.response[0]["raw_attributes"][self.cert_attr]:
            cert = x509.load_der_x509_certificate(raw)
            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):
                comment = f"{username}({cert.subject.rfc4514_string()})"
            key = serialize_ssh_public_key(cert.public_key()).decode()
            yield f"{key} {comment}"
    finally:
        conn.unbind()

only_auth_keys(username, cert)

Default filter: keep certs whose KeyUsage permits digital signature.

Source code in src/authkeys/sources/ldap.py 
25
26
27
28
29
30
31
32
33
def only_auth_keys(username: str, cert) -> bool:
    """Default filter: keep certs whose KeyUsage permits digital signature."""
    from cryptography import x509

    try:
        usage = cert.extensions.get_extension_for_class(x509.KeyUsage)
    except x509.ExtensionNotFound:
        return True
    return not usage or usage.value.digital_signature