globuzma/canaille-globuzma
1
0
Fork 0
forked from Github-Mirrors/canaille
Code Pull requests Activity

The LDAP Backend is now a class

Browse source
This commit is contained in:
Éloi Rivard 2023-04-08 20:42:38 +02:00
parent c27f0ccdab
commit 30282e633b
7 changed files with 204 additions and 160 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
canaille/__init__.py
View file

@ -174,11 +174,11 @@ def create_app(config=None, validate=True):
sentry_sdk = setup_sentry(app)
try:
from .oidc.oauth import setup_oauth
from .backends.ldap.backend import init_backend
from .backends.ldap.backend import LDAPBackend
from .app.i18n import setup_i18n
setup_logging(app)
init_backend(app)
LDAPBackend(app)
setup_oauth(app)
setup_blueprints(app)
setup_jinja(app)

14
canaille/app/commands.py
View file

@ -9,18 +9,16 @@ from flask.cli import with_appcontext
def with_backendcontext(func):
@functools.wraps(func)
def _func(*args, **kwargs):
from canaille.backends.ldap.backend import (
setup_backend,
teardown_backend,
)
from canaille.backends.ldap.backend import LDAPBackend
if not current_app.config["TESTING"]: # pragma: no cover
setup_backend(current_app)
backend = LDAPBackend(current_app)
backend.setup(current_app)
result = func(*args, **kwargs)
backend.teardown(current_app)
if not current_app.config["TESTING"]: # pragma: no cover
teardown_backend(current_app)
else:
result = func(*args, **kwargs)
return result

4
canaille/app/configuration.py
View file

@ -16,9 +16,9 @@ def validate(config, validate_remote=False):
if not validate_remote:
return
from canaille.backends.ldap.backend import validate_configuration
from canaille.backends.ldap.backend import LDAPBackend
validate_configuration(config)
LDAPBackend.validate(config)
validate_smtp_configuration(config)

37
canaille/backends/__init__.py
View file

@ -0,0 +1,37 @@
class Backend:
def __init__(self, app):
self.app = app
@self.app.before_request
def before_request():
if not app.config["TESTING"]:
return self.setup()
@self.app.after_request
def after_request(response):
if not app.config["TESTING"]:
self.teardown()
return response
def setup(self):
"""
This method will be called before each http request,
it should open the connection to the backend.
"""
raise NotImplementedError()
def teardown(self):
"""
This method will be called after each http request,
it should close the connections to the backend.
"""
raise NotImplementedError()
@classmethod
def validate(cls, config):
"""
This method should validate the config part dedicated to the backend.
It should raise :class:`~canaille.configuration.ConfigurationError` when
errors are met.
"""
raise NotImplementedError()

126
canaille/backends/ldap/backend.py
View file

@ -3,51 +3,19 @@ import uuid
import ldap
from canaille.app.configuration import ConfigurationException
from canaille.backends import Backend
from flask import g
from flask import render_template
from flask import request
from flask_babel import gettext as _
def setup_ldap_models(config):
from .ldapobject import LDAPObject
from canaille.core.models import Group
from canaille.core.models import User
class LDAPBackend(Backend):
def __init__(self, app):
setup_ldap_models(app.config)
super().__init__(app)
LDAPObject.root_dn = config["BACKENDS"]["LDAP"]["ROOT_DN"]
user_base = config["BACKENDS"]["LDAP"]["USER_BASE"].replace(
f',{config["BACKENDS"]["LDAP"]["ROOT_DN"]}', ""
)
User.base = user_base
User.rdn_attribute = config["BACKENDS"]["LDAP"].get(
"USER_ID_ATTRIBUTE", User.DEFAULT_ID_ATTRIBUTE
)
object_class = config["BACKENDS"]["LDAP"].get(
"USER_CLASS", User.DEFAULT_OBJECT_CLASS
)
User.ldap_object_class = (
object_class if isinstance(object_class, list) else [object_class]
)
group_base = (
config["BACKENDS"]["LDAP"]
.get("GROUP_BASE", "")
.replace(f',{config["BACKENDS"]["LDAP"]["ROOT_DN"]}', "")
)
Group.base = group_base or None
Group.rdn_attribute = config["BACKENDS"]["LDAP"].get(
"GROUP_ID_ATTRIBUTE", Group.DEFAULT_ID_ATTRIBUTE
)
object_class = config["BACKENDS"]["LDAP"].get(
"GROUP_CLASS", Group.DEFAULT_OBJECT_CLASS
)
Group.ldap_object_class = (
object_class if isinstance(object_class, list) else [object_class]
)
def setup_backend(app):
def setup(self):
try: # pragma: no cover
if request.endpoint == "static":
return
@ -55,18 +23,21 @@ def setup_backend(app):
pass
try:
g.ldap_connection = ldap.initialize(app.config["BACKENDS"]["LDAP"]["URI"])
g.ldap_connection = ldap.initialize(
self.app.config["BACKENDS"]["LDAP"]["URI"]
)
g.ldap_connection.set_option(
ldap.OPT_NETWORK_TIMEOUT, app.config["BACKENDS"]["LDAP"].get("TIMEOUT")
ldap.OPT_NETWORK_TIMEOUT,
self.app.config["BACKENDS"]["LDAP"].get("TIMEOUT"),
)
g.ldap_connection.simple_bind_s(
app.config["BACKENDS"]["LDAP"]["BIND_DN"],
app.config["BACKENDS"]["LDAP"]["BIND_PW"],
self.app.config["BACKENDS"]["LDAP"]["BIND_DN"],
self.app.config["BACKENDS"]["LDAP"]["BIND_PW"],
)
except ldap.SERVER_DOWN:
message = _("Could not connect to the LDAP server '{uri}'").format(
uri=app.config["BACKENDS"]["LDAP"]["URI"]
uri=self.app.config["BACKENDS"]["LDAP"]["URI"]
)
logging.error(message)
return (
@ -74,7 +45,7 @@ def setup_backend(app):
"error.html",
error=500,
icon="database",
debug=app.config.get("DEBUG", False),
debug=self.app.config.get("DEBUG", False),
description=message,
),
500,
@ -82,7 +53,7 @@ def setup_backend(app):
except ldap.INVALID_CREDENTIALS:
message = _("LDAP authentication failed with user '{user}'").format(
user=app.config["BACKENDS"]["LDAP"]["BIND_DN"]
user=self.app.config["BACKENDS"]["LDAP"]["BIND_DN"]
)
logging.error(message)
return (
@ -90,35 +61,19 @@ def setup_backend(app):
"error.html",
error=500,
icon="key",
debug=app.config.get("DEBUG", False),
debug=self.app.config.get("DEBUG", False),
description=message,
),
500,
)
def teardown_backend(app):
def teardown(self):
if g.get("ldap_connection"): # pragma: no branch
g.ldap_connection.unbind_s()
g.ldap_connection = None
def init_backend(app):
setup_ldap_models(app.config)
@app.before_request
def before_request():
if not app.config["TESTING"]:
return setup_backend(app)
@app.after_request
def after_request(response):
if not app.config["TESTING"]:
teardown_backend(app)
return response
def validate_configuration(config):
@classmethod
def validate(cls, config):
from canaille.core.models import Group
from canaille.core.models import User
@ -128,7 +83,8 @@ def validate_configuration(config):
ldap.OPT_NETWORK_TIMEOUT, config["BACKENDS"]["LDAP"].get("TIMEOUT")
)
conn.simple_bind_s(
config["BACKENDS"]["LDAP"]["BIND_DN"], config["BACKENDS"]["LDAP"]["BIND_PW"]
config["BACKENDS"]["LDAP"]["BIND_DN"],
config["BACKENDS"]["LDAP"]["BIND_PW"],
)
except ldap.SERVER_DOWN as exc:
@ -188,3 +144,41 @@ def validate_configuration(config):
user.delete(conn)
conn.unbind_s()
def setup_ldap_models(config):
from .ldapobject import LDAPObject
from canaille.core.models import Group
from canaille.core.models import User
LDAPObject.root_dn = config["BACKENDS"]["LDAP"]["ROOT_DN"]
user_base = config["BACKENDS"]["LDAP"]["USER_BASE"].replace(
f',{config["BACKENDS"]["LDAP"]["ROOT_DN"]}', ""
)
User.base = user_base
User.rdn_attribute = config["BACKENDS"]["LDAP"].get(
"USER_ID_ATTRIBUTE", User.DEFAULT_ID_ATTRIBUTE
)
object_class = config["BACKENDS"]["LDAP"].get(
"USER_CLASS", User.DEFAULT_OBJECT_CLASS
)
User.ldap_object_class = (
object_class if isinstance(object_class, list) else [object_class]
)
group_base = (
config["BACKENDS"]["LDAP"]
.get("GROUP_BASE", "")
.replace(f',{config["BACKENDS"]["LDAP"]["ROOT_DN"]}', "")
)
Group.base = group_base or None
Group.rdn_attribute = config["BACKENDS"]["LDAP"].get(
"GROUP_ID_ATTRIBUTE", Group.DEFAULT_ID_ATTRIBUTE
)
object_class = config["BACKENDS"]["LDAP"].get(
"GROUP_CLASS", Group.DEFAULT_OBJECT_CLASS
)
Group.ldap_object_class = (
object_class if isinstance(object_class, list) else [object_class]
)

9
demo/demoapp.py
View file

@ -14,15 +14,16 @@ def create_app():
@app.before_first_request
def populate():
from canaille.backends.ldap.backend import setup_backend
from canaille.backends.ldap.backend import teardown_backend
from canaille.backends.ldap.backend import LDAPBackend
from canaille.core.models import Group
from canaille.core.models import User
from canaille.core.populate import fake_groups
from canaille.core.populate import fake_users
from canaille.oidc.models import Client
setup_backend(app)
backend = LDAPBackend(app)
backend.setup()
jane = User(
formatted_name="Jane Doe",
given_name="Jane",
@ -141,6 +142,6 @@ def create_app():
fake_users(50)
fake_groups(10, nb_users_max=10)
teardown_backend(app)
backend.teardown()
return app

14
tests/backends/test_backends.py Normal file
View file

@ -0,0 +1,14 @@
import pytest
from canaille.backends import Backend
def test_required_methods(testclient):
with pytest.raises(NotImplementedError):
Backend.validate({})
backend = Backend(testclient.app)
with pytest.raises(NotImplementedError):
backend.setup()
with pytest.raises(NotImplementedError):
backend.teardown()