auth: Add LDAP support

Signed-off-by: Jan Lindemann <jan@janware.com>
This commit is contained in:
Jan Lindemann 2025-06-05 20:48:14 +02:00
commit 8a316ead21
4 changed files with 283 additions and 29 deletions

View file

@ -1,11 +1,13 @@
# -*- coding: utf-8 -*-
from typing import Optional, Union
from typing import Optional, Union, Self
import abc
from enum import Enum, auto
from jwutils import log, Config
from enum import Flag, Enum, auto
from jwutils.log import *
from jwutils import Config, load_object
class Access(Enum): # export
Read = auto()
@ -13,6 +15,11 @@ class Access(Enum): # export
Create = auto()
Delete = auto()
class ProjectFlags(Flag): # export
NoFlags = auto()
Contributing = auto()
Active = auto()
class Group: # export
def __repr__(self):
@ -33,40 +40,94 @@ class User: # export
@abc.abstractmethod
def _name(self) -> str:
pass
@abc.abstractmethod
def _groups(self) -> list[Group]:
pass
raise NotImplementedError
@property
def name(self) -> str:
return self._name()
def _display_name(self) -> str:
return self._name()
@property
def display_name(self) -> str:
return self._display_name()
@abc.abstractmethod
def _groups(self) -> list[Group]:
raise NotImplementedError
@property
def groups(self) -> list[Group]:
return self._groups()
class Auth: # export
@abc.abstractmethod
def _email(self) -> str:
raise NotImplementedError
@property
def email(self) -> str:
return self._email()
class Auth(abc.ABC): # export
@classmethod
def load(cls, tp: str, conf: Config) -> Self:
return load_object(f'jwutils.auth.{tp}.Auth', Auth, 'Auth', conf)
def __init__(self, conf: Config):
self.__conf = conf
@abc.abstractmethod
def _access(self, what: str, access_type: Optional[Access], who: User|Group|None) -> bool:
pass
@abc.abstractmethod
def _current_user(self) -> User:
pass
self.__base_user_by_email: Optional[dict[str, User]] = None
@property
def conf(self):
return self.__conf
@abc.abstractmethod
def _access(self, what: str, access_type: Optional[Access], who: User|Group|None) -> bool:
raise NotImplementedError
def access(self, what: str, access_type: Optional[Access]=None, who: Optional[Union[User|Group]]=None) -> bool:
return self._access(what, access_type, who)
@abc.abstractmethod
def _current_user(self) -> User:
raise NotImplementedError
@property
def current_user(self) -> User:
return self._current_user()
def access(self, what: str, access_type: Optional[Access]=None, who: Optional[Union[User|Group]]=None) -> bool:
return self._access(what, access_type, who)
@abc.abstractmethod
def _user(self, name) -> User:
raise NotImplementedError
def user(self, name) -> User:
return self._user(name)
@abc.abstractmethod
def _users(self) -> list[User]:
raise NotImplementedError
def _user_by_email(self, email: str) -> User:
if self.__base_user_by_email is None:
ret: dict[str, User] = dict()
users = self._users()
for user in users.values():
ret[user.email] = user
self.__base_user_by_email = ret
return self.__base_user_by_email[email]
def user_by_email(self, email) -> User:
return self._user_by_email(email)
@property
def users(self) -> list[User]:
return self._users()
@abc.abstractmethod
def _projects(self, name, flags: ProjectFlags) -> list[str]:
raise NotImplementedError
def projects(self, name, flags: ProjectFlags) -> list[str]:
return self._projects(name, flags)