config and ldap handling refactor

webapp/__init__.py

@@ -51,7 +51,7 @@ def start():
 
     app.connections = pools.LDAPConnectionPool(config.ldap_url, timeout=300.0)
     def drop_profile(dn):
-        if dn != config.admin_dn:
+        if dn != config.ldap_admin_dn:
             del app.profiles[dn]
     app.connections.register_callback('drop', drop_profile)
     app.connections.start()

webapp/admin.py

@@ -1,20 +1,31 @@
+import functools
 import ldap
 import re
 import flask
 
-from webapp import app, context, config
-from webapp.auth import login_required
+import webapp
+from webapp import app, context, config, ldaputils
 
 bp = flask.Blueprint('admin', __name__)
 
-def _ldap_not_in(patterns):
-    joined_patterns = ''.join(f'({p})' for p in patterns)
-    one_of_pattern = f'(|{joined_patterns})'
-    return f'!{one_of_pattern}'
+def admin_required_impl(f):
+    @functools.wraps(f)
+    def func(*a, **kw):
+        # TODO: Actually check for admin perms
+        if not flask.session['is_admin']:
+            flask.abort(403)
 
-def _ldap_get_users_list(conn, query='&'):
+        return f(*a, **kw)
+    return func
+
+def admin_required(f):
+    return webapp.auth.login_required(admin_required_impl(f))
+
+def _get_user_list(conn, query='&'):
+    """Returns List[Tuple[username, full name]] for query"""
     all_users = []
-    results = conn.search_s(config.ldap_people, ldap.SCOPE_SUBTREE, f'(&(uid=*)(cn=*)({query}))', attrlist=['uid', 'cn'])
+
+    results = conn.search_s(config.ldap_people, ldap.SCOPE_SUBTREE, f'(&(uid=*)(cn=*){ldaputils.wrap(query)})', attrlist=['uid', 'cn'])
     for user, attrs in results:
         user_uid = attrs['uid'][0].decode()
         user_cn = attrs['cn'][0].decode()
@@ -23,81 +34,71 @@ def _ldap_get_users_list(conn, query='&'):
     all_users.sort(key=lambda user: user[0].lower())
     return all_users
 
-def _ldap_get_all_users_groupped(conn):
-    group_queries = [
-        (group_name, f'memberOf={pattern}')
-        for group_name, pattern in config.admin_groups.items()
-    ]
-
+def _get_groupped_user_list(conn):
+    """Returns all users (uid, full name), groupped by active groups"""
     groupped_users = [
-        (group_name, _ldap_get_users_list(conn, query))
-        for group_name, query in group_queries
+        (group.capitalize(), _get_user_list(conn, f'memberOf={ldaputils.group_dn(group)}'))
+        for group in config.ldap_active_groups
     ]
 
-    other_users_query = _ldap_not_in(query for _, query in group_queries)
+    inactive_filter = ldaputils._not(
+        ldaputils.member_of_any(config.ldap_active_groups)
+    )
+
     groupped_users.append(
-        ('Other', _ldap_get_users_list(conn, other_users_query))
+        ('Inactive users', _get_user_list(conn, inactive_filter))
     )
 
     return groupped_users
 
 @bp.route('/admin/')
-@login_required
-def admin_list():
-    if not flask.session['is_admin']:
-        flask.abort(403)
+@admin_required
+def admin_view():
+    return flask.redirect('/admin/users/')
 
+@bp.route('/admin/users/')
+@admin_required
+def admin_users_view():
     conn = context.get_connection()
-    user_groups = _ldap_get_all_users_groupped(conn)
+    groups = _get_groupped_user_list(conn)
 
-    return flask.render_template('admin/list.html', user_groups=user_groups)
+    return flask.render_template('admin/list.html', groups=groups)
 
-def _ldap_get_user(conn, uid):
-    profile = []
+def _get_profile(conn, uid):
+    results = conn.search_s(ldaputils.user_dn(uid), ldap.SCOPE_SUBTREE)
+    return ldaputils.normalized_entries(results)[0]
 
-    for user, attrs in conn.search_s(config.dn_format % uid, ldap.SCOPE_SUBTREE):
-        for attr, values in attrs.items():
-            for value in values:
-                profile.append((attr, value.decode()))
-
-    return profile
-
-def _rendered_ldap_profile(profile):
-    rendered_profile = []
-    for attr, value in profile:
+def _format_profile(profile):
+    rendered = []
+    dn, attrs = profile
+    for attr, value in attrs:
         attr_sanitized = attr.lower()
         attr_full_name = config.full_name.get(attr_sanitized, attr_sanitized)
         attr_readable_name = config.readable_names.get(attr_full_name)
-        rendered_profile.append((attr, attr_readable_name, value))
+        rendered.append((attr, attr_readable_name, value))
 
     # Attributes with readable names first, then by name
-    rendered_profile.sort(key=lambda profile: profile[0])
-    rendered_profile.sort(key=lambda profile: profile[1] is None)
-    return rendered_profile
+    rendered.sort(key=lambda profile: profile[0])
+    rendered.sort(key=lambda profile: profile[1] is None)
+    return rendered
 
-def _ldap_get_user_groups(conn, uid):
-    groups = []
-    user_dn = config.dn_format % uid
-    filter = f'(&(objectClass=groupOfUniqueNames)(uniqueMember={user_dn}))'
-    for group_dn, attrs in conn.search_s(config.ldap_base, ldap.SCOPE_SUBTREE, filter):
-        groups.append(attrs['cn'][0].decode())
+def _get_groups_of(conn, uid):
+    filter = ldaputils.groups_of_user(uid)
+    groups = [
+        attrs['cn'][0].decode()
+        for group_dn, attrs in
+        conn.search_s(config.ldap_base, ldap.SCOPE_SUBTREE, filter)
+    ]
 
     return groups
 
-def _ldap_validate_uid(uid):
-    if not re.match(r'^[a-zA-Z_][a-zA-Z0-9-_]*\Z', uid):
-        raise RuntimeError('Invalid uid')
-
 @bp.route('/admin/users/<uid>')
-@login_required
+@admin_required
 def admin_user_view(uid):
-    if not flask.session['is_admin']:
-        flask.abort(403)
-
+    ldaputils.validate_name(uid)
     conn = context.get_connection()
-    _ldap_validate_uid(uid)
 
-    profile = _ldap_get_user(conn, uid)
-    groups = _ldap_get_user_groups(conn, uid)
+    profile = _get_profile(conn, uid)
+    groups = _get_groups_of(conn, uid)
 
-    return flask.render_template('admin/user.html', uid=uid, profile=_rendered_ldap_profile(profile), groups=groups)
+    return flask.render_template('admin/user.html', uid=uid, profile=_format_profile(profile), groups=groups)

webapp/auth.py

@@ -3,7 +3,7 @@ import ldap
 import flask
 import urllib
 
-from webapp import app, context, config
+from webapp import app, context, config, ldaputils
 
 bp = flask.Blueprint('auth', __name__)
 
@@ -39,7 +39,7 @@ def login_action():
     username = flask.request.form.get("username", "").lower()
     password = flask.request.form.get("password", "")
     goto = flask.request.values.get("goto", "/")
-    dn = config.dn_format % username
+    dn = ldaputils.user_dn(username)
 
     conn = _connect_to_ldap(dn, password)
     if conn:
@@ -51,7 +51,7 @@ def login_action():
                 username = vs[0].decode()
 
         # Check if user belongs to admin group
-        is_admin = bool(conn.search_s(dn, ldap.SCOPE_SUBTREE, f'memberOf={config.ldapweb_admin_group}'))
+        is_admin = bool(conn.search_s(dn, ldap.SCOPE_SUBTREE, ldaputils.member_of_any(config.ldap_admin_groups)))
 
         flask.session["username"] = username
         flask.session['dn'] = dn

webapp/config.py.dist

@@ -1,6 +1,7 @@
 import flask_wtf
 import wtforms
 import secrets
+import os
 
 hackerspace_name = 'Warsaw Hackerspace'
 secret_key = secrets.token_hex(32)
@@ -12,20 +13,20 @@ kadmin_principal_map = "{}@HACKERSPACE.PL"
 # LDAP configuration
 
 ldap_url = 'ldap://ldap.hackerspace.pl'
-dn_format = "uid=%s,ou=people,dc=hackerspace,dc=pl"
-
-ldapweb_admin_group = 'cn=ldap-admin,ou=Group,dc=hackerspace,dc=pl'
-
 ldap_base = 'dc=hackerspace,dc=pl'
-ldap_people = 'ou=People,dc=hackerspace,dc=pl'
-admin_groups = {
-    'Fatty': 'cn=fatty,ou=Group,dc=hackerspace,dc=pl',
-    'Starving': 'cn=starving,ou=Group,dc=hackerspace,dc=pl',
-    'Potato': 'cn=potato,ou=Group,dc=hackerspace,dc=pl',
-}
+ldap_people = 'ou=people,dc=hackerspace,dc=pl'
+ldap_user_dn_format = 'uid={},ou=people,dc=hackerspace,dc=pl'
+ldap_group_dn_format = 'cn={},ou=group,dc=hackerspace,dc=pl'
 
-admin_dn = 'cn=ldapweb,ou=Services,dc=hackerspace,dc=pl'
-admin_pw = 'changeme'
+# user groups allowed to see /admin
+ldap_admin_groups = os.getenv('LDAPWEB_ADMIN_GROUPS', 'ldap-admin,staff,zarzad').split(',')
+
+# user groups indicating that a user is active
+ldap_active_groups = os.getenv('LDAPWEB_ACTIVE_GROUPS', 'fatty,starving,potato').split(',')
+
+# service user with admin privileges (for admin listings, creating new users)
+ldap_admin_dn = os.getenv('LDAPWEB_ADMIN_DN', 'cn=ldapweb,ou=services,dc=hackerspace,dc=pl')
+ldap_admin_password = os.getenv('LDAPWEB_ADMIN_PASSWORD', 'unused')
 
 # LDAP attribute configuration
 

webapp/context.py

@@ -25,9 +25,9 @@ def get_connection(dn = None):
     return app.connections[dn]
 
 def get_admin_connection():
-    conn = app.connections[config.admin_dn]
+    conn = app.connections[config.ldap_admin_dn]
     if not conn:
-        conn = app.connections.bind(config.admin_dn, config.admin_pw)
+        conn = app.connections.bind(config.ldap_admin_dn, config.ldap_admin_password)
     return conn
 
 def get_profile():

webapp/ldaputils.py

@@ -0,0 +1,66 @@
+import re
+import ldap
+from webapp import config
+
+def is_valid_name(name):
+  """`true` if `name` is a safe ldap uid/cn"""
+  return re.match(r'^[a-zA-Z_][a-zA-Z0-9-_]*\Z', name) is not None
+
+def validate_name(name):
+  """Raises `RuntimeError` if `name` is not a safe ldap uid/cn"""
+  if not is_valid_name(name):
+    raise RuntimeError('Invalid name')
+
+def user_dn(uid):
+  validate_name(uid)
+  return config.ldap_user_dn_format.format(uid)
+
+def group_dn(cn):
+  validate_name(cn)
+  return config.ldap_group_dn_format.format(cn)
+
+def wrap(filter):
+  if len(filter) and filter[0] == '(' and filter[-1] == ')':
+    return filter
+  else:
+    return f'({filter})'
+
+def _or(*filters):
+  wrapped = ''.join(wrap(f) for f in filters)
+  return f'(|{wrapped})'
+
+def _and(*filters):
+  wrapped = ''.join(wrap(f) for f in filters)
+  return f'(&{wrapped})'
+
+def _not(filter):
+  wrapped = wrap(filter)
+  return f'(!{wrapped})'
+
+def member_of_any(groups):
+  """Returns a filter that matches users that are a member of any of the given group names"""
+  return _or(*(f'memberOf={group_dn(group)}' for group in groups))
+
+def groups_of_user(uid):
+  """Returns a filter that matches groups that have the given user as a member"""
+  return f'(&(objectClass=groupOfUniqueNames)(uniqueMember={user_dn(uid)}))'
+
+def normalized_entries(entries):
+  """
+  Converts ldap entries from python-ldap format into a more convenient
+  List[Tuple[
+    dn,
+    List[tuple[attr_name, attr_value]]
+  ]]
+  """
+  normalized = []
+  for dn, attrs in entries:
+    normalized_attrs = []
+    for attr, values in attrs.items():
+      for value in values:
+        normalized_attrs.append((attr, value.decode()))
+    normalized.append((dn, normalized_attrs))
+
+  return normalized
+
+

webapp/pools.py

@@ -7,8 +7,7 @@ class LDAPConnectionPool(lru.LRUPool):
         lru.LRUPool.__init__(self, **kw)
         self.use_tls = use_tls
         self.url = url
-        self.admin_dn = config.admin_dn
-        self.admin_pw = config.admin_pw
+
     @lru.locked
     def bind(self, dn, password):
         conn = ldap.initialize(self.url)

webapp/templates/admin/list.html

@@ -2,7 +2,7 @@
 {% block content %}
 <p>Good evening, professor {{ session['username'] }}. All LDAP accounts:</p>
 
-{% for group_name, users in user_groups %}
+{% for group_name, users in groups %}
   <h2>{{ group_name }}</h2>
   <ul>
     {% for uid, name in users %}