#!/usr/bin/python3
# SPDX-FileCopyrightText: 2004-2026 Univention GmbH
# SPDX-License-Identifier: AGPL-3.0-only
"""|UDM| access to handler modules."""
from __future__ import annotations
import copy
import importlib
import locale
import os
import warnings
from importlib import reload as reload_module
from typing import TYPE_CHECKING, Any, Protocol, overload
import ldap
from ldap.filter import filter_format
import univention.admin
import univention.admin.handlers
import univention.admin.hook
import univention.admin.syntax
import univention.admin.uldap
from univention.admin import localization
from univention.admin._ucr import configRegistry
from univention.admin.layout import Group, ILayoutElement, Tab
from univention.admin.log import log
if TYPE_CHECKING:
from univention.admin.handlers import _Attributes, simpleLdap
[docs]
class UdmModule(Protocol):
module: str = ''
childs: bool = False
childmodules: list[str] = []
operations: list[str] = []
short_description: str = ''
object_name: str = ''
object_name_plural: str = ''
long_description: str = ''
options: dict[str, univention.admin.option] = {}
property_descriptions: dict[str, univention.admin.property] = {}
default_property_descriptions: dict[str, univention.admin.property] = {}
policy_apply_to: list[str] = []
policy_position_dn_prefix: str = ''
policy_oc: str = ''
docleanup: bool = False
layout: list[Tab] = []
mapping: univention.admin.mapping.mapping = None
initialized: bool = False
extended_attribute_tabnames: list[str] = []
extended_udm_attributes: list[univention.admin.extended_attribute] = []
[docs]
class object:
def __init__(self, co: None, lo: univention.admin.uldap.access, position: univention.admin.uldap.position, dn: str = '', superordinate: simpleLdap | None = None, attributes: _Attributes | None = None) -> None:
pass
[docs]
@staticmethod
def identify(dn: str, attr: dict[str, list[Any]]) -> bool:
pass
[docs]
@staticmethod
def lookup(co: None, lo: univention.admin.uldap.access, filter: str = '', base: str = '', superordinate: Any = None, scope: str = 'base+one', unique: bool = False, required: bool = False, timeout: int = -1, sizelimit: int = 0, authz: bool = True) -> list[Any]:
pass
[docs]
@staticmethod
def lookup_filter(filter_s: str | None = None, lo: univention.admin.uldap.access | None = None) -> univention.admin.filter.conjunction:
pass
UdmName = UdmModule | str
log = log.getChild(__name__)
translation = localization.translation('univention/admin')
_ = translation.translate
modules: dict[str, UdmModule] = {}
"""Mapping from module name to Python module."""
_superordinates: set[str] = set()
"""List of all module names (strings) that are _superordinates."""
containers: list[UdmModule] = []
@univention.admin._ldap_cache(ttl=3600)
def _ldap_operational_attribute_names(lo: univention.admin.uldap.access) -> set[str]:
schema = lo.authz_connection.get_schema()
attrs = [schema.get_obj(ldap.schema.models.AttributeType, x) for x in schema.listall(ldap.schema.models.AttributeType)]
usages = [ldap.schema.models.AttributeUsage[o] for o in ('directoryoperation', 'dsaoperation', 'distributedoperation')]
return {n.lower() for a in attrs for n in a.names if a.usage in usages}
ORDERING_BY_MATCHING_RULE = {
# Equality
'caseIgnoreMatch': 'caseIgnoreOrderingMatch',
'caseExactMatch': 'caseExactOrderingMatch',
'integerMatch': 'integerOrderingMatch',
'generalizedTimeMatch': 'generalizedTimeOrderingMatch',
'UUIDMatch': 'UUIDOrderingMatch',
'uuidMatch': 'UUIDOrderingMatch',
'CSNMatch': 'CSNOrderingMatch',
# Substring
'caseIgnoreSubstringsMatch': 'caseIgnoreOrderingMatch',
'caseExactSubstringsMatch': 'caseExactOrderingMatch',
'numericStringSubstringsMatch': 'numericStringOrderingMatch',
'octetStringSubstringsMatch': 'octetStringOrderingMatch',
}
ORDERING_BY_SYNTAX = {
'1.3.6.1.4.1.1466.115.121.1.15': 'caseIgnoreOrderingMatch', # Directory String
'1.3.6.1.4.1.1466.115.121.1.27': 'integerOrderingMatch', # Integer
'1.3.6.1.4.1.1466.115.121.1.24': 'generalizedTimeOrderingMatch', # Generalized Time
}
@univention.admin._ldap_cache(ttl=3600)
def _ldap_ordering_rules(lo: univention.admin.uldap.access) -> dict[str, str]:
"""
Get a heuristic mapping of lowercased attribute names and their possible ORDERING matching rules.
For Server Side Sorting we need to specify an ORDERING matching rule for the attributes.
But most attributes don't specify it in the schema, which enforces the client to specify one explicitly.
We use this best-guess-from-schema-introspection instead of forcing all UDM module properties to specify them, adding a lot of redundancy
or problems for third-party modules not being extended yet.
"""
schema = lo.authz_connection.get_schema()
attr_cls = ldap.schema.models.AttributeType
def first(value):
if isinstance(value, (tuple, list)):
return value[0] if value else None
return value
def effective(attr, field):
seen = set()
while attr:
value = first(getattr(attr, field, None))
if value is not None:
return value
sup = first(getattr(attr, 'sup', None))
if not sup or sup in seen:
return None
seen.add(sup)
attr = schema.get_obj(attr_cls, sup)
return None
ordering_rules = {}
for oid in schema.listall(attr_cls):
attr = schema.get_obj(attr_cls, oid)
if not attr:
continue
# if schema already defines ORDERING, no heuristic is needed.
if effective(attr, 'ordering'):
continue
equality = effective(attr, 'equality')
substr = effective(attr, 'substr')
syntax = effective(attr, 'syntax')
ordering = (
ORDERING_BY_MATCHING_RULE.get(equality)
or ORDERING_BY_MATCHING_RULE.get(substr)
or ORDERING_BY_SYNTAX.get(syntax)
)
if not ordering:
continue
for name in attr.names:
ordering_rules[name.lower()] = ordering
return ordering_rules
[docs]
def update() -> None:
"""Scan file system and update internal list of |UDM| handler modules."""
global modules, _superordinates
_modules: dict[str, UdmModule] = {}
superordinates: set[str] = set()
# since last update(), syntax.d and hooks.d may have changed (Bug #31154)
univention.admin.syntax.import_syntax_files()
univention.admin.hook.import_hook_files()
def _walk(root: str, dir: str, files: list[str]) -> None:
for file in files:
if not file.endswith('.py') or file.startswith('__'):
continue
package = os.path.join(dir, file)[len(root) + 1:-len('.py')]
log.debug('importing module', type=package)
modulepackage = '.'.join(package.split(os.path.sep))
m: Any = importlib.import_module('univention.admin.handlers.%s' % (modulepackage,))
m.initialized = False
if not hasattr(m, 'module'):
log.error('importing module: attribute "module" is missing', type=modulepackage)
continue
_modules[m.module] = m
if isContainer(m):
containers.append(m)
superordinates.update(superordinate_names(m))
for root in univention.admin.handlers.__path__: # type: ignore
for w_root, _w_dirs, w_files in os.walk(root):
_walk(root, w_root, w_files)
modules = _modules
_superordinates = superordinates
# since last update(), syntax.d may have new choices
# put here as one syntax wants to provide all modules
univention.admin.syntax.update_choices()
@overload
def get(module: UdmModule) -> UdmModule:
pass
@overload
def get(module: str) -> UdmModule | None:
pass
[docs]
def get(module: UdmName) -> UdmModule | None:
"""
Get |UDM| module.
:param module: either the name (str) of a module or the module itself.
:returns: the module or `None` if no module exists with the requested name.
"""
# FIXME: raise Exception instead of returning None
if not module:
return None # type: ignore
if isinstance(module, str):
return modules.get(module) # type: ignore
return module # type: ignore[return-value]
def _get(module: UdmModule | str) -> UdmModule:
"""
Internal function to lazy-load and get |UDM| module.
:param module: either the name (str) of a module or the module itself.
:returns: the module
:raises KeyError: if no module exists with the requested name.
"""
if not modules:
update()
if isinstance(module, str):
return modules[module]
return module
[docs]
def init(lo: univention.admin.uldap.access, position: univention.admin.uldap.position, module: UdmModule, template_object: simpleLdap | None = None, force_reload: bool = False) -> None:
"""
Initialize |UDM| handler module.
:param lo: |LDAP| connection.
:param position: |UDM| position instance.
:param module: |UDM| handler module.
:param template_object: Reference to a instance, from which the default values are used.
:param force_reload: With `True` force Python to reload the module from the file system.
"""
if isinstance(lo, univention.uldap.access):
log.error('Wrong access class in use! Use univention.admin.uldap instead of univention.uldap!', stack_info=True)
warnings.warn('Wrong access class in use! Use univention.admin.uldap instead of univention.uldap!', DeprecationWarning, stacklevel=3)
if configRegistry.is_true('directory/manager/type-checking/strict'):
raise TypeError('Expect univention.admin.uldap.access!')
# you better do a reload if init is called a second time
# especially because update_extended_attributes
# called twice will have side-effects
if force_reload:
reload_module(module) # type: ignore
if module.module == 'users/user':
# users/self inherits from users/user leading to errors when not reloading it as well
# TODO: remove when droppng Python2.7 support and all super(object, self) calls have been replaced with super()
reload_module(univention.admin.modules._get('users/self')) # type: ignore
# reset property descriptions to defaults if possible
if hasattr(module, 'default_property_descriptions'):
module.property_descriptions = copy.deepcopy(module.default_property_descriptions)
# log.debug('initialize module: reset default descriptions')
# overwrite property descriptions
univention.admin.ucr_overwrite_properties(module, lo)
# check for properties with the syntax class LDAP_Search
for pname, prop in module.property_descriptions.items():
if prop.syntax.name == 'LDAP_Search':
prop.syntax._load(lo)
if prop.syntax.viewonly:
module.mapping.unregister(pname, False)
elif univention.admin.syntax.is_syntax(prop.syntax, univention.admin.syntax.complex) and hasattr(prop.syntax, 'subsyntaxes'):
for _text, subsyn in prop.syntax.subsyntaxes:
if subsyn.name == 'LDAP_Search':
subsyn._load(lo)
# add new properties, only for modules in default ldap base
if lo.compare_dn(configRegistry['ldap/base'].lower(), getattr(module.object, 'ldap_base', configRegistry['ldap/base']).lower()):
update_extended_options(lo, module, position)
update_extended_attributes(lo, module, position)
# get defaults from template
if template_object:
log.debug('initialize module with template', template=template_object.dn)
template_object.open()
# add template defaults
for key, tmpl in template_object.items():
if key == '_options':
if tmpl not in ([''], []):
for option in module.options.keys():
module.options[option].default = option in tmpl
elif key not in {'name', 'description', 'univentionObjectIdentifier'}: # these keys are part of the template itself
module.property_descriptions[key].base_default = copy.copy(tmpl)
module.property_descriptions[key].templates.append(template_object)
log.debug('initialize module: module.property_description after template', properties=module.property_descriptions)
# re-build layout if there any overwrites defined
univention.admin.ucr_overwrite_module_layout(module)
# some choices depend on extended_options/attributes
univention.admin.syntax.update_choices()
module.initialized = True
[docs]
def update_extended_options(lo: univention.admin.uldap.access, module: UdmModule, position: univention.admin.uldap.position) -> None:
"""Overwrite options defined via |LDAP|."""
# get current language
lang = locale.getlocale(locale.LC_MESSAGES)[0]
log.trace('update extended_options', lang=lang)
module_filter = filter_format('(univentionUDMOptionModule=%s)', [name(module)])
if name(module) == 'settings/usertemplate':
module_filter = '(|(univentionUDMOptionModule=users/user)%s)' % (module_filter,)
# append UDM extended options
new_options = copy.copy(module.options) if hasattr(module, 'options') else {}
for _dn, attrs in lo.authz_connection.search(base=position.getDomainConfigBase(), filter='(&(objectClass=univentionUDMOption)%s)' % (module_filter,)):
oname = attrs['cn'][0].decode('UTF-8', 'replace')
shortdesc = _get_translation(lang, attrs, 'univentionUDMOptionTranslationShortDescription;entry-%s', 'univentionUDMOptionShortDescription')
longdesc = _get_translation(lang, attrs, 'univentionUDMOptionTranslationLongDescription;entry-%s', 'univentionUDMOptionLongDescription')
default = attrs.get('univentionUDMOptionDefault', [b'0'])[0] == b'1'
editable = attrs.get('univentionUDMOptionEditable', [b'0'])[0] == b'1'
classes = [x.decode('UTF-8', 'replace') for x in attrs.get('univentionUDMOptionObjectClass', [])]
is_app_option = attrs.get('univentionUDMOptionIsApp', [b'0'])[0] == b'1'
new_options[oname] = univention.admin.option(
short_description=shortdesc,
long_description=longdesc,
default=default,
editable=editable,
objectClasses=classes,
is_app_option=is_app_option,
)
module.options = new_options
[docs]
class EA_Layout(dict):
"""Extended attribute layout."""
def __init__(self, **kwargs):
dict.__init__(self, kwargs)
@property
def name(self) -> str:
return self.get('name', '')
@property
def overwrite(self) -> str | None:
return self.get('overwrite', None)
@property
def tabName(self) -> str:
return self.get('tabName', '')
@property
def groupName(self) -> str:
return self.get('groupName', '')
@property
def position(self) -> int:
return self.get('position', -1)
@property
def groupPosition(self) -> int:
return self.get('groupPosition', -1)
@property
def advanced(self) -> bool:
return self.get('advanced', False)
@property
def is_app_tab(self) -> bool:
return self.get('is_app_tab', False)
def __lt__(self, other):
return (self.groupName, self.position, self.name) < (other.groupName, other.position, other.name)
def __gt__(self, other):
return (self.groupName, self.position, self.name) > (other.groupName, other.position, other.name)
def __eq__(self, other):
return (self.groupName, self.position, self.name) == (other.groupName, other.position, other.name)
def __le__(self, other):
return (self.groupName, self.position, self.name) <= (other.groupName, other.position, other.name)
def __ge__(self, other):
return (self.groupName, self.position, self.name) >= (other.groupName, other.position, other.name)
def __ne__(self, other):
return (self.groupName, self.position, self.name) != (other.groupName, other.position, other.name)
def __hash__(self):
return hash((self.groupName, self.position, self.name))
[docs]
def update_extended_attributes(lo: univention.admin.uldap.access, module: UdmModule, position: univention.admin.uldap.position) -> None:
"""Load extended attribute from |LDAP| and modify |UDM| handler."""
# add list of tabnames created by extended attributes (in case module is initialized a second time and EAs were removed)
if not hasattr(module, 'extended_attribute_tabnames'):
module.extended_attribute_tabnames = []
# append UDM extended attributes
properties4tabs: dict[str, list[EA_Layout]] = {}
tab_positions: dict[str, int] = {}
group_positions: dict[tuple[str, str], int] = {}
overwrite_tabs: set[str] = set()
module.extended_udm_attributes = []
lang = locale.getlocale(locale.LC_MESSAGES)[0]
def ldap_true(attrs, attribute_name: str) -> bool:
return attrs.get(attribute_name, [b'0'])[0].upper() in {b'1', b'TRUE'}
def ldap_int(attrs, attribute_name, default=0):
try:
return int(attrs.get(attribute_name, [default])[0])
except ValueError:
return default
def merge_position(current: int | None, candidate: int, *, label: str, _log) -> int:
if candidate < 1:
return current if current is not None else -1
if current is None or current < 1:
return candidate
if current != candidate:
_log.warning(
'Conflicting extended attribute layout positions; using the lowest position',
label=label,
positions=sorted({current, candidate}),
)
return min(current, candidate)
def layout_sort_key(position: int, label: str) -> tuple[bool, int, str]:
"""Sort positioned layout elements first and use their label as a stable tie-breaker."""
return position < 1, position if position > 0 else 0, label
modname = name(module)
module_filter = filter_format('(univentionUDMPropertyModule=%s)', [modname])
if modname == 'settings/usertemplate':
module_filter = '(|(univentionUDMPropertyModule=users/user)%s)' % (module_filter,)
new_property_descriptions = copy.copy(module.property_descriptions)
for _dn, attrs in lo.authz_connection.search(
base=position.getDomainConfigBase(), filter='(&(objectClass=univentionUDMProperty)%s(univentionUDMPropertyVersion=2))' % (module_filter,),
):
_log = log.bind(dn=_dn, type=modname)
# get CLI name
pname = attrs['univentionUDMPropertyCLIName'][0].decode('UTF-8', 'replace')
object_class = attrs.get('univentionUDMPropertyObjectClass', [])[0].decode('UTF-8', 'replace')
ldap_attribute_name = attrs['univentionUDMPropertyLdapMapping'][0].decode('UTF-8', 'replace')
if modname == 'settings/usertemplate' and object_class == 'univentionMail' and b'settings/usertemplate' not in attrs.get('univentionUDMPropertyModule', []):
continue # since "mail" is a default option, creating a usertemplate with any mail attribute would raise Object class violation: object class 'univentionMail' requires attribute 'uid'
_log.trace('update extended_attributes', lang=str(lang), attributes=attrs)
# get syntax
propertySyntaxString = attrs.get('univentionUDMPropertySyntax', [b''])[0].decode('utf-8', 'replace')
if propertySyntaxString and hasattr(univention.admin.syntax, propertySyntaxString):
propertySyntax = getattr(univention.admin.syntax, propertySyntaxString)
else:
if lo.authz_connection.searchDn(filter=filter_format(univention.admin.syntax.LDAP_Search.FILTER_PATTERN, [propertySyntaxString])):
propertySyntax = univention.admin.syntax.LDAP_Search(propertySyntaxString)
else:
propertySyntax = univention.admin.syntax.string()
# get hooks
propertyHookString = attrs.get('univentionUDMPropertyHook', [b''])[0].decode('utf-8', 'replace')
propertyHook = None
if propertyHookString and hasattr(univention.admin.hook, propertyHookString):
propertyHook = getattr(univention.admin.hook, propertyHookString)()
# get default value
propertyDefault = [x.decode('UTF-8') if x is not None else x for x in attrs.get('univentionUDMPropertyDefault', [None])]
# value may change
mayChange = ldap_true(attrs, 'univentionUDMPropertyValueMayChange')
# prevent UMC default popup
preventUmcDefaultPopup = ldap_true(attrs, 'univentionUDMPropertyPreventUmcDefaultPopup')
# value is editable (only via hooks or direkt module.info[] access)
editable = not ldap_true(attrs, 'univentionUDMPropertyValueNotEditable')
copyable = not ldap_true(attrs, 'univentionUDMPropertyCopyable') # yes, broken inverted logic, Bug #56172
# value is required
valueRequired = ldap_true(attrs, 'univentionUDMPropertyValueRequired')
# value not available for searching
doNotSearch = ldap_true(attrs, 'univentionUDMPropertyDoNotSearch')
# check if EA is multivalue property
multivalue = ldap_true(attrs, 'univentionUDMPropertyMultivalue')
map_method = propertyHook.map if propertyHook else None
unmap_method = propertyHook.unmap if propertyHook else None
if not multivalue:
is_default_unmap = propertyHook and propertyHook.__class__.unmap is univention.admin.hook.simpleHook.unmap
unmap_method = propertyHook.unmap if propertyHook and not is_default_unmap else univention.admin.mapping.ListToString
if propertySyntaxString == 'boolean' and (not propertyHook or is_default_unmap):
unmap_method = univention.admin.mapping.BooleanListToString
map_method = univention.admin.mapping.BooleanMap
# single value ==> use only first value
propertyDefault = propertyDefault[0]
# Show this attribute in UDM/UMC?
layoutDisabled = ldap_true(attrs, 'univentionUDMPropertyLayoutDisable')
# get descriptions
shortdesc = _get_translation(lang, attrs, 'univentionUDMPropertyTranslationShortDescription;entry-%s', 'univentionUDMPropertyShortDescription')
longdesc = _get_translation(lang, attrs, 'univentionUDMPropertyTranslationLongDescription;entry-%s', 'univentionUDMPropertyLongDescription')
# create property
fullWidth = ldap_true(attrs, 'univentionUDMPropertyLayoutFullWidth')
new_property_descriptions[pname] = univention.admin.property(
short_description=shortdesc,
long_description=longdesc,
syntax=propertySyntax,
multivalue=multivalue,
options=[x.decode('UTF-8', 'replace') for x in attrs.get('univentionUDMPropertyOptions', [])],
required=valueRequired,
may_change=mayChange,
prevent_umc_default_popup=preventUmcDefaultPopup,
dontsearch=doNotSearch,
default=propertyDefault,
editable=editable,
copyable=copyable,
size='Two' if fullWidth else None,
)
# add LDAP mapping
if ldap_attribute_name.lower() != 'objectclass':
module.mapping.register(pname, ldap_attribute_name, map_method, unmap_method)
else:
module.mapping.register(pname, ldap_attribute_name, univention.admin.mapping.nothing, univention.admin.mapping.nothing)
deleteObjectClass = ldap_true(attrs, 'univentionUDMPropertyDeleteObjectClass')
module.extended_udm_attributes.append(
univention.admin.extended_attribute(
name=pname,
objClass=object_class,
ldapMapping=ldap_attribute_name,
deleteObjClass=deleteObjectClass,
syntax=propertySyntaxString,
hook=propertyHook,
),
)
if ldap_attribute_name.lower() in _ldap_operational_attribute_names(lo):
module.object._static_ldap_attributes.add(ldap_attribute_name)
if layoutDisabled:
for tab in getattr(module, 'layout', []):
tab.remove(pname)
if layoutDisabled or not hasattr(module, 'layout'):
continue
tabname = _get_translation(lang, attrs, 'univentionUDMPropertyTranslationTabName;entry-%s', 'univentionUDMPropertyLayoutTabName', _('Custom'))
overwriteTab = ldap_true(attrs, 'univentionUDMPropertyLayoutOverwriteTab')
# in the first generation of extended attributes of version 2
# this field was a position defining the attribute to
# overwrite. now it is the name of the attribute to overwrite
overwriteProp: str | None = attrs.get('univentionUDMPropertyLayoutOverwritePosition', [b''])[0].decode('UTF-8', 'replace')
if overwriteProp == '0':
overwriteProp = None
tabAdvanced = ldap_true(attrs, 'univentionUDMPropertyLayoutTabAdvanced')
groupname = _get_translation(lang, attrs, 'univentionUDMPropertyTranslationGroupName;entry-%s', 'univentionUDMPropertyLayoutGroupName')
# This number specifies the position on which this group is placed on the tab. The numbering starts at 1
groupPosition = ldap_int(attrs, 'univentionUDMPropertyLayoutGroupPosition', -1)
# This number specifies the position on which this extended attributes is placed on the tab or in the group. The numeration starts at 1
prop_position = ldap_int(attrs, 'univentionUDMPropertyLayoutPosition', -1)
tabPosition = -1 # one day, could be specifyable
# only one is possible ==> overwriteTab wins
if overwriteTab and overwriteProp:
overwriteProp = None
# add tab name to list if missing
if tabname not in properties4tabs:
properties4tabs[tabname] = []
_log.trace('update extended_attributes: custom fields init for tab', tab=tabname)
# remember tab for purging if required
if overwriteTab:
overwrite_tabs.add(tabname)
tab_positions[tabname] = merge_position(
tab_positions.get(tabname),
tabPosition,
label=tabname,
_log=_log,
)
if groupname:
group_key = (tabname, groupname)
group_positions[group_key] = merge_position(
group_positions.get(group_key),
groupPosition,
label='%s/%s' % group_key,
_log=_log,
)
properties4tabs[tabname].append(
EA_Layout(
name=pname,
tabName=tabname,
position=max(prop_position, -1),
advanced=tabAdvanced,
overwrite=overwriteProp,
fullWidth=fullWidth,
groupName=groupname,
groupPosition=groupPosition,
is_app_tab=any(
option in [key for (key, value) in getattr(module, 'options', {}).items() if value.is_app_option]
for option in attrs.get('univentionUDMPropertyOptions', [])
),
),
)
module.property_descriptions = new_property_descriptions
# overwrite tabs that have been added by UDM extended attributes
tabs_to_clear = overwrite_tabs | set(module.extended_attribute_tabnames)
if properties4tabs:
# remove layout of tabs that have been marked for replacement
for tab in module.layout:
if tab.label in tabs_to_clear:
tab.layout = []
sorted_tabnames = sorted(
properties4tabs,
key=lambda tabname: layout_sort_key(tab_positions.get(tabname, -1), tabname),
)
next_tab_index = 0
for tabname in sorted_tabnames:
priofields = sorted(
properties4tabs[tabname],
key=lambda ea_layout: (
layout_sort_key(group_positions.get((tabname, ea_layout.groupName), -1), ea_layout.groupName),
layout_sort_key(ea_layout.position, ea_layout.name),
),
)
currentTab = None
# get existing fields if tab has not been overwritten
for tab in module.layout:
if tab.label == tabname:
# found tab in layout
currentTab = tab
break
else:
# tab not found in current layout, so add it at its configured position
currentTab = Tab(tabname, tabname, advanced=True)
tab_position = tab_positions.get(tabname, -1)
if tab_position > 0:
# keep positioned EA tabs in sorted order when their requested
# insertion positions overlap due to earlier insertions.
index = min(max(tab_position - 1, next_tab_index), len(module.layout))
module.layout.insert(index, currentTab)
next_tab_index = index + 1
else:
module.layout.append(currentTab)
# remember tabs that have been added by UDM extended attributes
if tabname not in module.extended_attribute_tabnames:
module.extended_attribute_tabnames.append(tabname)
currentTab.is_app_tab = any(x.is_app_tab for x in priofields)
# check if tab is empty ==> overwritePosition is impossible
freshTab = not currentTab.layout
# create every group exactly once
groupnames = sorted(
{ea_layout.groupName for ea_layout in priofields if ea_layout.groupName},
key=lambda groupname: layout_sort_key(group_positions.get((tabname, groupname), -1), groupname),
)
next_group_index = 0
for groupname in groupnames:
if any(isinstance(item, ILayoutElement) and item.label == groupname for item in currentTab.layout):
continue
grp = Group(groupname)
group_position = group_positions.get((tabname, groupname), -1)
if group_position > 0:
# keep positioned EA groups in sorted order when their requested
# insertion positions overlap due to earlier insertions.
index = min(max(group_position - 1, next_group_index), len(currentTab.layout))
currentTab.layout.insert(index, grp)
next_group_index = index + 1
else:
currentTab.layout.append(grp)
for ea_layout in priofields:
if currentTab.advanced and not ea_layout.advanced:
currentTab.advanced = False
# - existing property shall be overwritten AND
# - tab is not new and has not been cleaned before AND
# - position >= 1 (top left position is defined as 1) AND
# - old property with given position exists
if currentTab.exists(ea_layout.name):
continue
elif ea_layout.overwrite and not freshTab: # we want to overwrite an existing property
# in the global fields ...
if not ea_layout.groupName:
replaced, _layout = currentTab.replace(ea_layout.overwrite, ea_layout.name, recursive=True)
if not replaced: # the property was not found so we'll append it
currentTab.layout.append(ea_layout.name)
else:
for item in currentTab.layout:
if isinstance(item, ILayoutElement) and item.label == ea_layout.groupName:
replaced, _layout = item.replace(ea_layout.overwrite, ea_layout.name)
if not replaced: # the property was not found so we'll append it
item.layout.append(ea_layout.name)
break
else:
if not ea_layout.groupName:
currentTab.insert(ea_layout.position, ea_layout.name)
else:
for item in currentTab.layout:
if isinstance(item, ILayoutElement) and item.label == ea_layout.groupName:
item.insert(ea_layout.position, ea_layout.name)
break
# check for properties with the syntax class LDAP_Search
for pname, prop in module.property_descriptions.items():
if prop.syntax.name == 'LDAP_Search':
prop.syntax._load(lo)
if prop.syntax.viewonly:
module.mapping.unregister(pname, False)
elif univention.admin.syntax.is_syntax(prop.syntax, univention.admin.syntax.complex) and hasattr(prop.syntax, 'subsyntaxes'):
for _text, subsyn in prop.syntax.subsyntaxes:
if subsyn.name == 'LDAP_Search':
subsyn._load(lo)
[docs]
def identify(dn: str, attr: dict[str, list[Any]], module_name: str = '', canonical: int = 0, module_base: str | None = None) -> list[UdmModule]:
"""
Return list of |UDM| handlers capable of handling the given |LDAP| object.
:param dn: |DN| of the |LDAP| object.
:param attr: |LDAP| attributes.
:param module_name: If given only the given module name is used if it is capable to handle the object.
:param canonical: UNUSED!
:param module_base: Optional string the module names must start with.
:returns: the list of |UDM| modules.
"""
res = [m for m in (modules.get(mt.decode('ASCII', 'replace')) for mt in attr.get('univentionObjectType', [])) if m]
if not res:
for name, module in modules.items():
if module_base is not None and not name.startswith(module_base):
continue
if not hasattr(module, 'identify'):
log.warning('module does not provide identify', type=name)
continue
if (not module_name or module_name == module.module) and module.identify(dn, attr):
res.append(module)
if not res:
log.debug('object could not be identified', dn=dn)
for r in res:
log.trace('identifies object', type=r.module, dn=dn)
return res
[docs]
def identifyOne(dn: str, attr: dict[str, list[Any]], type: str = '') -> UdmModule | None:
"""
Return the |UDM| handler capable of handling the given |LDAP| object.
:param dn: |DN| of the |LDAP| object.
:param atr: |LDAP| attributes.
:param type: If given only the given module name is used if it is capable to handle the object.
:returns: the |UDM| modules or `None`.
"""
res = identify(dn, attr, type)
if len(res) != 1:
return None
else:
return res[0]
[docs]
def recognize(module_name: str, dn: str, attr: dict[str, list[Any]]) -> bool:
module = get(module_name)
if not hasattr(module, 'identify'):
return False
return module.identify(dn, attr)
[docs]
def name(module: UdmName) -> str:
"""Return name of module."""
if not module:
return ''
return get(module).module
[docs]
def superordinate_names(module_name: UdmName) -> list[str]:
"""Return name of superordinate module."""
module = get(module_name)
names = getattr(module, 'superordinate', [])
if isinstance(names, str):
names = [names]
return names
[docs]
def superordinate_name(module_name: UdmName) -> str | None:
"""
Return name of first superordinate module.
.. deprecated :: UCS 4.2
Use :py:func:`superordinate_names` instead.
"""
names = superordinate_names(module_name)
return names[0] if names else None
[docs]
def superordinate(module: UdmModule) -> UdmModule:
"""
Return instance of superordinate module.
.. deprecated :: UCS 4.2
Use :py:func:`superordinates` instead.
"""
return get(superordinate_name(module))
[docs]
def superordinates(module: UdmName) -> list[UdmModule | None]:
"""Return instance of superordinate module."""
return [get(x) for x in superordinate_names(module)]
[docs]
def subordinates(module: UdmName) -> list[UdmModule]:
"""
Return list of instances of subordinate modules.
:param module: ???
:returns: list of |UDM| handler modules.
"""
return [mod for mod in modules.values() if name(module) in superordinate_names(mod) and not isContainer(mod)]
[docs]
def find_superordinate(dn: str, co: None, lo: univention.admin.uldap.access) -> UdmModule | None:
"""
For a given |DN|, search in the |LDAP| path whether this LDAP object is
below an object that is a superordinate or is a superordinate itself.
:param dn: |DN|.
:param co: |UDM| configuation object.
:param lo: |LDAP| connection.
:returns: the superordinate module or `None`.
"""
# walk up the ldap path and stop if we find an object type that is a superordinate
while dn:
attr = lo.authz_connection.get(dn) # TODO: information disclosure?
module = identifyOne(dn, attr)
if module and isSuperordinate(module):
return get(module)
dn = lo.parentDn(dn)
return None
[docs]
def options(module_name: UdmName) -> dict[str, Any]:
"""return options available for module"""
module = get(module_name)
return getattr(module, 'options', {})
[docs]
def attributes(module_name: UdmName) -> list[dict[str, str]]:
"""
Return attributes for module.
:param module_name: the name of the |UDM| module, e.g. `users/user`.
"""
module = get(module_name)
return [
{'name': attribute, 'description': module.property_descriptions[attribute].short_description}
for attribute in module.property_descriptions.keys()
]
[docs]
def short_description(module_name: UdmName) -> str:
"""
Return short description for module.
:param module_name: the name of the |UDM| module, e.g. `users/user`.
:returns: The short descriptive text.
"""
module = get(module_name)
if hasattr(module, 'short_description'):
return module.short_description
modname = name(module)
if modname:
return modname
return repr(module)
[docs]
def policy_short_description(module_name: UdmName) -> str:
"""
Return short description for policy module primarily used for tab headers.
:param module_name: the name of the |UDM| policy module, e.g. `policies/pwhistory`.
:returns: The short descriptive text.
"""
module = get(module_name)
return getattr(module, 'policy_short_description', short_description(module))
[docs]
def long_description(module_name: UdmName) -> str:
"""
Return long description for module.
:param module_name: the name of the |UDM| module, e.g. `users/user`.
:returns: The long descriptive text.
"""
module = get(module_name)
return getattr(module, 'long_description', short_description(module))
[docs]
def childs(module_name: UdmName) -> bool:
"""
Return whether module may have subordinate modules.
:param module_name: the name of the |UDM| module, e.g. `users/user`.
:returns: `True` if the module has children, `False` otherwise.
"""
module = get(module_name)
return getattr(module, 'childs', False)
[docs]
def virtual(module_name: UdmName) -> bool:
"""
Return whether the module is virtual (alias for other modules).
:param module_name: the name of the |UDM| module, e.g. `computers/computer`.
:returns: `True` if the module is virtual, `False` otherwise.
"""
module = get(module_name)
return getattr(module, 'virtual', False)
[docs]
def lookup(module_name: UdmName, co: None, lo: univention.admin.uldap.access, filter: str = '', base: str = '', superordinate: Any = None, scope: str = 'base+one', unique: bool = False, required: bool = False, timeout: int = -1, sizelimit: int = 0, authz: bool = True) -> list[Any]:
"""
Return objects of module that match the given criteria.
:param module_name: the name of the |UDM| module, e.g. `users/user`.
"""
module = get(module_name)
tmpres = []
if hasattr(module, 'lookup'):
kw = {} if authz else {'authz': authz} # TODO: not every UDM module supports the new parameter
tmpres = module.lookup(co, lo, filter, base=base, superordinate=superordinate, scope=scope, unique=unique, required=required, timeout=timeout, sizelimit=sizelimit, **kw)
# check for 'None' items just in case...
return [item for item in tmpres if item]
[docs]
def isSuperordinate(module: UdmName) -> bool:
"""
Check if the module is a |UDM| superordinate module.
:param module: A |UDM| handler class.
:returns: `True` if the handler is a superordinate module, `False` otherwise.
"""
return name(module) in _superordinates
[docs]
def isContainer(module: UdmModule) -> bool:
"""
Check if the module is a |UDM| container module.
:param module: A |UDM| handler class.
:returns: `True` if the handler is a container module, `False` otherwise.
"""
return name(module).startswith('container/')
[docs]
def isPolicy(module: UdmModule) -> bool:
"""
Check if the module is a |UDM| policy module.
:param module: A |UDM| handler class.
:returns: `True` if the handler is a policy module, `False` otherwise.
"""
return name(module).startswith('policies/')
[docs]
def defaultPosition(module: UdmModule, superordinate: Any = None) -> str:
"""
Returns default position for object of module.
:param module: A |UDM| handler class.
:param superordinate: A optional superordinate |UDM| object instance.
:returns: The |DN| of the container for the object.
"""
rdns = ['users', 'dns', 'dhcp', 'shares', 'printers']
base = univention.admin.uldap.getBaseDN()
if superordinate:
return superordinate.dn
start = name(module).split('/')[0]
if start in rdns:
return 'cn=%s,%s' % (ldap.dn.escape_dn_chars(start), base)
return base
[docs]
def supports(module_name: str, operation: str) -> bool:
"""
Check if module supports operation
:param module_name: the name of the |UDM| module, e.g. `users/user`.
:param operation: the name of the operation, e.g. 'edit'.
:returns: `True` if the operation is supported, `False` otherwise.
"""
module = get(module_name)
if not hasattr(module, 'operations'):
return True
return operation in module.operations
[docs]
def objectType(co: None, lo: univention.admin.uldap.access, dn: str, attr: _Attributes | None = None, modules: list[UdmModule] = [], module_base: str | None = None) -> list[str]:
if not dn:
return []
if attr is None:
attr = lo.authz_connection.get(dn) # TODO: information disclosure?
if not attr:
return []
ot = attr.get('univentionObjectType')
if ot:
return [x.decode('utf-8') for x in ot]
if not modules:
modules = identify(dn, attr, module_base=module_base)
return [name(mod) for mod in modules]
[docs]
def objectShadowType(co: None, lo: univention.admin.uldap.access, dn: str, attr: _Attributes | None = None, modules: list[UdmModule] = []) -> list[Any]:
# FIXME: This returns a nested list[...list[str]] for containers!
return [
objectShadowType(co, lo, lo.parentDn(dn)) if otype and otype.startswith('container/') else otype
for otype in objectType(co, lo, dn, attr, modules)
]
[docs]
def findObject(co: None, lo: univention.admin.uldap.access, dn: str, type: UdmModule, attr: _Attributes | None = None, module_base: str | None = None) -> Any | None:
if attr is None:
attr = lo.authz_connection.get(dn) # TODO: information disclosure?
if not attr:
return None
ndn = dn
nattr = attr
while True:
for module in identify(ndn, nattr):
if module and module.module == type:
s = superordinate(module)
so = findObject(co, lo, ndn, s) if s else None
return module.object(co, lo, ndn, superordinate=so)
ndn = lo.parentDn(ndn)
if not ndn:
break
nattr = lo.authz_connection.get(ndn) # TODO: information disclosure?
return None
[docs]
def policyOc(module_name: UdmName) -> str:
"""
Return the |LDAP| objectClass used to store the policy.
:param module_name: the name of the |UDM| policy module, e.g. `policies/pwhistory`.
:returns: the objectClass.
"""
module = get(module_name)
return getattr(module, 'policy_oc', '')
[docs]
def policiesGroup(module_name: UdmName) -> str:
"""
Return the name of the group the |UDM| policy belongs to.
:param module_name: the name of the |UDM| policy module, e.g. `policies/pwhistory`.
:returns: the group name.
"""
module = get(module_name)
return getattr(module, 'policies_group', 'top')
[docs]
def policies() -> list[univention.admin.policiesGroup]:
res: dict[str, list[str]] = {}
for mod in modules.values():
if not isPolicy(mod):
continue
if name(mod) != 'policies/policy':
res.setdefault(policiesGroup(mod), []).append(name(mod))
return [
univention.admin.policiesGroup(id=groupname, members=sorted(members))
for groupname, members in sorted(res.items())
]
[docs]
def policyTypes(module_name: str) -> list[str]:
"""
Returns a list of policy types applying to the given module.
:param module_name: the name of the |UDM| module, e.g. `users/user`.
:returns: a list of |UDM| policy modules, e.g. `policies/pwhistory`.
"""
if not module_name:
return []
if module_name not in modules:
return []
return [
name
for name, module in modules.items()
if name.startswith('policies/') and module_name in getattr(module, 'policy_apply_to', ())
]
[docs]
def policyPositionDnPrefix(module_name: UdmName) -> str:
"""
Return the relative |DN| for a policy.
:param module_name: the name of the |UDM| policy module, e.g. `policies/pwhistory`.
:return: A |DN| string to append to the |LDAP| base to get the container for the policy.
"""
module = get(module_name)
if not hasattr(module, 'policy_position_dn_prefix'):
return ''
policy_position_dn_prefix = module.policy_position_dn_prefix
policy_position_dn_prefix = policy_position_dn_prefix.removesuffix(',')
return policy_position_dn_prefix
[docs]
def defaultContainers(module: simpleLdap) -> list[str]:
"""
Checks for the attribute default_containers that should contain a
list of RDNs of default containers.
:param module: |UDM|
:returns: a list of DNs.
"""
base = getattr(module.object, 'ldap_base', configRegistry['ldap/base'])
return ['%s,%s' % (rdn, base) for rdn in getattr(module, 'default_containers', [])]
[docs]
def childModules(module_name: UdmName) -> list[str]:
"""
Return child modules if module is a super module.
:param module_name: the name of the |UDM| module, e.g. `users/user`.
:returns: List of child module names.
"""
module = get(module_name)
return list(getattr(module, 'childmodules', []))
def _get_translation(locale: str | None, attrs: Any, name: str, defaultname: str, default: str = '') -> str:
if locale:
locale = locale.replace('_', '-').lower()
if name % (locale,) in attrs:
return attrs[name % (locale,)][0].decode('UTF-8', 'replace')
locale = locale.split('-', 1)[0]
name_short_lang = name % (locale,)
if name_short_lang in attrs:
return attrs[name_short_lang][0].decode('UTF-8', 'replace')
for key in attrs:
if key.startswith(name_short_lang):
return attrs[key][0].decode('UTF-8', 'replace')
return attrs.get(defaultname, [default.encode('utf-8')])[0].decode('UTF-8', 'replace')
