Publishes a user's public key to Active Directory, generating a new key pair in the process if the user doesn't have one.

publish_ssh_key_to_ad.py

SERVER_URI = 'dc.corp.local'                # Active Directory Domain Controller IP/FQDN
BASE_DN = 'OU=LAB,DC=corp,DC=local'         # Organizational Unit to start user search in
DOMAIN = 'corp.local'                       # AD domain name
SSH_KEY_ATTR = 'sshPublicKeys'              # AD user object attribute for storing the keys 

def cleanup(conn):
    conn.unbind()

def generate_keypair():
    from cryptography.hazmat.primitives.asymmetric import rsa
    from cryptography.hazmat.primitives import serialization
    private_key = rsa.generate_private_key(
        public_exponent = 65537,
        key_size = 3072
    )
    private_pem = private_key.private_bytes(
        encoding = serialization.Encoding.PEM,
        format = serialization.PrivateFormat.OpenSSH,
        encryption_algorithm = serialization.NoEncryption()
    ).decode("utf-8")
    print(f"Generated private key:\n\n{private_pem}\n\nSave this to '~/.ssh/id_rsa'. YOU WILL NOT SEE IT AGAIN.)

    public_key = private_key.public_key()
    public_pem = public_key.public_bytes(
        encoding = serialization.Encoding.OpenSSH,
        format = serialization.PublicFormat.OpenSSH
    ).decode("utf-8")
    return public_pem

def main():
    from ldap3 import Server, Connection, MODIFY_ADD
    from getpass import getpass
    
    username = input('LDAP Username: ')
    password = getpass('LDAP Password: ')
    ssh_key = input('SSH Public Key (blank to generate new keypair): ')
    if (ssh_key == ''):
        ssh_key = generate_keypair()

    filter = f"(sAMAccountName={username})"
    server = Server(SERVER_URI)
    conn = Connection(server, user=f"{username}@{DOMAIN}", password=password)
    if not (conn.bind()):
        print(f"[ERROR] Unable to bind LDAP! LastError: {conn.last_error}")
        cleanup(conn)

    if not (conn.search(BASE_DN, filter, attributes=['sAMAccountName', SSH_KEY_ATTR])):
        print("[ERROR] Unable to find user!")
        cleanup(conn)

    user = conn.entries[0]
    if (conn.modify(user.entry_dn, {SSH_KEY_ATTR: [(MODIFY_ADD, [ssh_key])]})):
        print("Successfully updated SSH public key.")
        cleanup(conn)
    else:
        print(f"[ERROR] Attempt to update public key failed with result: {conn.last_error}")
        cleanup(conn)

if __name__ == "__main__":
    main()