121 lines
5.3 KiB
Python
Executable File
121 lines
5.3 KiB
Python
Executable File
"""Explicit, administrator-controlled service grants for each effective level."""
|
||
|
||
import re
|
||
|
||
from django.core.exceptions import ImproperlyConfigured
|
||
from ldap3.core.exceptions import LDAPInvalidDnError
|
||
from ldap3.utils.dn import parse_dn
|
||
|
||
|
||
SERVICE_LINKS = {
|
||
"Discourse · community forums": {
|
||
"url": "https://discourse.eom.dev/",
|
||
"favicon": "https://discourse.eom.dev/uploads/default/optimized/1X/_129430568242d1b7f853bb13ebea28b3f6af4e7_2_32x32.png",
|
||
},
|
||
"Gitea · Git hosting": {
|
||
"url": "https://gitea.eom.dev/",
|
||
"favicon": "https://gitea.eom.dev/assets/img/favicon.svg",
|
||
},
|
||
"Matrix · community chat": {
|
||
"url": "https://element.eom.dev/",
|
||
"favicon": "https://element.eom.dev/vector-icons/144.5a63bf2.png",
|
||
},
|
||
"Mastodon · social networking": {
|
||
"url": "https://mastodon.eom.dev/",
|
||
"favicon": "https://mastodon.eom.dev/packs/assets/favicon-48x48-DMnduFKh.png",
|
||
},
|
||
"PeerTube · video sharing": {
|
||
"url": "https://peertube.eom.dev/",
|
||
"favicon": "https://peertube.eom.dev/client/assets/images/favicon.png",
|
||
},
|
||
}
|
||
|
||
|
||
def validate_service_groups(value, paid_levels):
|
||
if not isinstance(value, dict):
|
||
raise ImproperlyConfigured("Membership service groups must be an object mapping levels to lists of LDAP group DNs.")
|
||
valid_levels = {0, *paid_levels}
|
||
for level, groups in value.items():
|
||
if not isinstance(level, str) or not re.fullmatch(r"0|[1-9][0-9]*", level) or int(level) not in valid_levels:
|
||
raise ImproperlyConfigured("Service group levels must be 0 or a configured paid membership level.")
|
||
if not isinstance(groups, list):
|
||
raise ImproperlyConfigured("Each service group level must contain a list of LDAP group DNs.")
|
||
seen = set()
|
||
for dn in groups:
|
||
if not isinstance(dn, str) or not dn or dn != dn.strip() or len(dn) > 1024 or any(ord(char) < 32 for char in dn):
|
||
raise ImproperlyConfigured("Service group DNs must be nonempty valid LDAP DNs of at most 1024 characters.")
|
||
try:
|
||
parts = parse_dn(dn)
|
||
except LDAPInvalidDnError:
|
||
raise ImproperlyConfigured("Invalid LDAP service group DN.") from None
|
||
if not parts or dn.casefold() in seen:
|
||
raise ImproperlyConfigured("Service group DNs must be valid and unique within each level.")
|
||
seen.add(dn.casefold())
|
||
return value
|
||
|
||
|
||
def configured_groups(tiers, service_groups):
|
||
"""All managed targets, not only the groups this account currently earns."""
|
||
return {tier["group_dn"] for tier in tiers.values()} | {
|
||
dn for groups in service_groups.values() for dn in groups
|
||
}
|
||
|
||
|
||
def service_names(level, service_groups):
|
||
"""Public labels from the same exact-level mapping used by reconciliation."""
|
||
labels = {
|
||
"dex users": "Dex · single sign-on",
|
||
"discourse users": "Discourse · community forums",
|
||
"gitea users": "Gitea · Git hosting",
|
||
"matrix users": "Matrix · community chat",
|
||
"mastodon users": "Mastodon · social networking",
|
||
"peertube users": "PeerTube · video sharing",
|
||
}
|
||
names = []
|
||
for dn in service_groups.get(str(level), []):
|
||
# Show only a group's common name, never its directory hierarchy.
|
||
common_name = next((value for attribute, value, _separator in parse_dn(dn) if attribute.casefold() == "cn"), "")
|
||
if common_name.casefold() == "dex users":
|
||
continue
|
||
if not common_name:
|
||
name = "Additional community service"
|
||
else:
|
||
# LDAP DN values escape reserved characters and UTF-8 bytes.
|
||
raw = re.sub(r"\\([,=+<>#;\"\\ ])", r"\1", common_name)
|
||
raw = re.sub(r"(?:\\[0-9a-fA-F]{2})+", lambda match: bytes.fromhex(match.group().replace("\\", "")).decode("utf-8", errors="replace"), raw)
|
||
name = labels.get(raw.casefold(), re.sub(r"\s+users$", "", raw, flags=re.IGNORECASE))
|
||
if name not in names:
|
||
names.append(name)
|
||
return names
|
||
|
||
|
||
def service_access_catalog(tiers, service_groups):
|
||
"""List each advertised service once, with the levels that include it."""
|
||
level_labels = {0: "Open Lab"}
|
||
level_labels.update({tier["level"]: tier["label"] for tier in tiers.values()})
|
||
levels = sorted(level_labels)
|
||
service_levels = {}
|
||
for level in levels:
|
||
for name in service_names(level, service_groups):
|
||
service_levels.setdefault(name, []).append(level)
|
||
|
||
catalog = []
|
||
for name, included_levels in service_levels.items():
|
||
if included_levels == levels:
|
||
access = "Every membership level"
|
||
else:
|
||
ranges = []
|
||
start = previous = included_levels[0]
|
||
for level in included_levels[1:]:
|
||
if level != previous + 1:
|
||
ranges.append((start, previous))
|
||
start = level
|
||
previous = level
|
||
ranges.append((start, previous))
|
||
access = ", ".join(
|
||
level_labels[first] if first == last else f"{level_labels[first]}–{level_labels[last]}"
|
||
for first, last in ranges
|
||
)
|
||
catalog.append({"name": name, "access": access, **SERVICE_LINKS.get(name, {})})
|
||
return catalog
|