hswaw/machines/customs.hackerspace.pl/openvpn-auth/openvpn_auth/__init__.py

import ldap3
import os
import sys
import ssl
from ldap3.utils.conv import escape_filter_chars

class NotActiveMember(Exception):
    "Person is not an active hackerspace member"

def check_member(uid: str, password: str):
    escaped_uid = escape_filter_chars(uid)
    user_dn = f"uid={escaped_uid},ou=People,dc=hackerspace,dc=pl"

    tls_configuration = ldap3.Tls(validate=ssl.CERT_REQUIRED, version=ssl.PROTOCOL_TLSv1)
    server = ldap3.Server("ldap.hackerspace.pl", use_ssl=True, tls=tls_configuration)
    with ldap3.Connection(server, user=user_dn, password=password, raise_exceptions=True) as conn:
        filterstr = (
            "(&"
                f"(uid={escaped_uid})"
                "(objectClass=hsMember)"
                "(|"
                    "(memberOf=cn=starving,ou=Group,dc=hackerspace,dc=pl)"
                    "(memberOf=cn=fatty,ou=Group,dc=hackerspace,dc=pl)"
                    "(memberOf=cn=potato,ou=Group,dc=hackerspace,dc=pl)"
                ")"
            ")")
        conn.search('ou=People,dc=hackerspace,dc=pl',
            filterstr,
            search_scope = ldap3.LEVEL,
            attributes = ['uid'])
        for e in conn.entries:
            if e['uid'] == uid:
                break
        else:
            NotActiveMember(f'Member {uid} not found in active members groups')

def member_auth():
    import argparse
    import getpass

    uid = os.environ.get('username', None)
    password = os.environ.get('password', None)
    
    if uid is None and password is None:
        print('"username" and "password" not found in environment')
        parser = argparse.ArgumentParser()
        parser.add_argument("uid", nargs='?', default=getpass.getuser(), help="user id")
        args = parser.parse_args()

        uid = args.uid
        password = getpass.getpass()
    
    try:
        check_member(uid, password)
        sys.exit(0)
    except Exception:
        sys.exit(1)

if __name__ == "__main__":
    member_auth()