from flask import current_app from ldap3_orm import AttrDef, EntryBase as _EntryBase, ObjectDef, EntryType from ldap3_orm import Reader from ldap3 import Entry, HASHED_SALTED_SHA256 from ldap3.utils.conv import escape_filter_chars from ldap3.utils.hashed import hashed from flask_login import UserMixin from ldap3.core.exceptions import LDAPSessionTerminatedByServerError from cryptography.hazmat.primitives import hashes from cryptography.hazmat.primitives import serialization from collections.abc import MutableSequence from datetime import datetime from dateutil import tz import pyotp import json import logging import crypt from flask_sqlalchemy import SQLAlchemy, orm from datetime import datetime import uuid import pyotp logger = logging.getLogger(__name__) ldap_conn = None # type: Connection base_dn = '' db = SQLAlchemy() # type: SQLAlchemy class SecurityUser(UserMixin): def __init__(self, username): self._username = username def get_id(self): return self._username class LambdaStr: def __init__(self, lam): self.lam = lam def __str__(self): return self.lam() class EntryBase(db.Model): __abstract__ = True # for sqlalchemy _type = None # will get replaced by the local type _query_object = None # will get replaced by the local type _base_dn = LambdaStr(lambda: base_dn) # def __init__(self, ldap_object=None, **kwargs): # if ldap_object is None: # self._ldap_object = self.get_type()(**kwargs) # else: # self._ldap_object = ldap_object def __str__(self): return str(self._ldap_object) @classmethod def get_object_def(cls): return ObjectDef(cls.object_classes, ldap_conn) @classmethod def get_base(cls): return cls.base_dn.format(_base_dn=base_dn) @classmethod def get_type(cls): if cls._type is None: cls._type = EntryType(cls.dn.replace('{base_dn}',cls.get_base()), cls.object_classes, ldap_conn) return cls._type def commit(self): self._ldap_object.entry_commit_changes() def add(self): print(self._ldap_object.entry_attributes_as_dict) ret = ldap_conn.add( self.dn, self.object_classes, self._ldap_object.entry_attributes_as_dict) logger.debug(ret) pass @classmethod def query_(cls): if cls._query_object is None: cls._query_object = cls._query(cls) return cls._query_object class _query(object): def __init__(self, clazz): self._class = clazz def _mapping(self, ldap_object): return ldap_object def _query(self, ldap_filter: str): reader = Reader(ldap_conn, self._class.get_object_def(), self._class.get_base(), ldap_filter) try: reader.search() except LDAPSessionTerminatedByServerError: ldap_conn.bind() reader.search() return [self._mapping(entry) for entry in reader] def all(self): return self._query(None) class Service(object): def __init__(self, name): self._name = name self._client_cert = False self._pki_config = { 'cn': '{username}', 'email': '{username}@{domain}' } @staticmethod def from_config(name, config): """ """ service = Service(name) if 'client_cert' in config: service._client_cert = bool(config['client_cert']) if 'pki_config' in config: service._pki_config.update(config['pki_config']) return service @property def name(self): return self._name @property def client_cert(self): return self._client_cert @property def pki_config(self): if not self._client_cert: raise Exception('invalid call') return self._pki_config class Certificate(object): def __init__(self, cn, ca_name: str, cert_data, revoked=False): self._cn = cn self._ca_name = ca_name self._cert_data = cert_data self._revoked = revoked self._cert_data.not_valid_after.replace(tzinfo=tz.tzutc()) self._cert_data.not_valid_before.replace(tzinfo=tz.tzutc()) @property def cn(self): return self._cn @property def ca_name(self): return self._ca_name @property def not_valid_before(self) -> datetime: return self._cert_data.not_valid_before.replace(tzinfo=tz.tzutc()).astimezone(tz.tzlocal()).replace(tzinfo=None) @property def not_valid_after(self) -> datetime: return self._cert_data.not_valid_after.replace(tzinfo=tz.tzutc()).astimezone(tz.tzlocal()).replace(tzinfo=None) @property def serial_number(self) -> int: return self._cert_data.serial_number @property def serial_number_hex(self) -> str: return f'{self._cert_data.serial_number:X}' def fingerprint(self, algorithm=hashes.SHA256()) -> bytes: return self._cert_data.fingerprint(algorithm) @property def is_valid(self) -> bool: return self.not_valid_after > datetime.now() and not self._revoked def pem(self) -> str: return self._cert_data.public_bytes(encoding=serialization.Encoding.PEM).decode() @property def raw(self): return self._cert_data def __str__(self): return f'Certificate(cn={self._cn}, ca_name={self._ca_name}, not_valid_before={self.not_valid_before}, not_valid_after={self.not_valid_after})' def generate_uuid(): return str(uuid.uuid4()) class User(EntryBase): id = db.Column( db.String(length=36), primary_key=True, default=generate_uuid) username = db.Column( db.String, unique=True) totps = db.relationship('Totp', back_populates='user') dn = "uid={uid},{base_dn}" base_dn = "ou=users,{_base_dn}" object_classes = ["top", "inetOrgPerson", "LenticularUser"] def __init__(self,**kwargs): self._ldap_object = None super(db.Model).__init__(**kwargs) @property def is_authenticated(self): return True # TODO def get(self, key): print(f'getitem: {key}') def make_writeable(self): self._ldap_object = self._ldap_object.entry_writable() @property def entry_dn(self): return self._ldap_object.entry_dn @property def fullname(self): return self._ldap_object.fullname @property def givenname(self): return self._ldap_object.givenname @property def surname(self): return self._ldap_object.surname @property def email(self): return self._ldap_object.mail @property def alternative_email(self): return self._ldap_object.altMail @property def auth_role(self): return self._ldap_object.authRole @property def gpg_public_key(self): return self._ldap_object.gpgPublicKey def change_password(self, password_new: str): self.make_writeable() password_hashed = crypt.crypt(password_new) self._ldap_object.userPassword = ('{CRYPT}' + password_hashed).encode() self.commit() class _query(EntryBase._query): def _mapping(self, ldap_object): user = User.query.filter(User.username == str(ldap_object.uid)).first() if user is None: # migration time user = User() user.username = str(ldap_object.uid) db.session.add(user) db.session.commit() user._ldap_object = ldap_object return user def by_username(self, username) -> 'User': result = self._query('(uid={username:s})'.format(username=escape_filter_chars(username))) if len(result) > 0: return result[0] else: return None class Totp(db.Model): id = db.Column(db.Integer, primary_key=True) secret = db.Column(db.String, nullable=False) name = db.Column(db.String, nullable=False) created_at = db.Column(db.DateTime, default=datetime.now, nullable=False) user_id = db.Column( db.String(length=36), db.ForeignKey(User.id), nullable=False) user = db.relationship(User) def verify(self, token: str): totp = pyotp.TOTP(self.secret) return totp.verify(token) class Group(EntryBase): __abstract__ = True # for sqlalchemy, disable for now dn = "cn={cn},{base_dn}" base_dn = "ou=Users,{_base_dn}" object_classes = ["top"] fullname = AttrDef("cn") class UserSignUp(db.Model): id = db.Column(db.Integer, primary_key=True) username = db.Column(db.String, nullable=False) password = db.Column(db.String, nullable=False) alternative_email = db.Column(db.String)