univention.lib package

Contents

univention.lib package#

Univention Lib Python module.

exception univention.lib.I18N_Error[source]#

Bases: Exception

Error in Internationalization.

class univention.lib.Locale(locale: str | None = None)[source]#

Bases: object

Represents a locale specification and provides simple access to language, territory, codeset and modifier.

Parameters:

locale (str or None) – The locale string language[_territory][.codeset][@modifier].

>>> Locale("deu_GER")
>>> str(Locale("ca_ES@valencia"))
>>> str(Locale(""))
REGEX = re.compile('^(?P<language>([a-z]{2}|C|POSIX))(?:_(?P<territory>[A-Z]{2}))?(?:\\.(?P<codeset>[a-zA-Z-0-9]+))?(?:@(?P<modifier>.+))?$')#
parse(locale: str) None[source]#

Parse locale string.

Parameters:

locale (str) – The locale string language[_territory][.codeset][@modifier].

Raises:
class univention.lib.NullTranslation(namespace: str, locale_spec: str | None = None, localedir: str | None = None)[source]#

Bases: object

Dummy translation.

Parameters:
  • namespace (str) – The name of the translation domain.

  • locale_spec (str) – The selected locale.

  • localedir (str) – The name of the directory containing the translation files.

property domain#
property locale: Locale | None#

Return currently selected locale.

Returns:

The currently selected locale.

Return type:

Locale

set_language(language: str = '') None[source]#

Select language.

Parameters:

language (str) – The language code.

translate(message: str) str[source]#

Translate message.

Parameters:

message (str) – The message to translate.

Returns:

The localized message.

Return type:

str

class univention.lib.Translation(*args, **kwargs)[source]#

Bases: NullTranslation

Translation.

locale: Locale = <univention.lib.i18n.Locale object>#
classmethod set_all_languages(language: str) None[source]#

Set the language of all existing Translation instances. This is required when instances are created during import time but later on the language should be changed.

set_language(language: str = '') None[source]#

Select language.

Parameters:

language (str) – The language code.

Raises:

I18N_Error – if the given locale is not valid.

univention.lib.get_lock(name: str, nonblocking: bool = False) IO[str] | None[source]#

Get a exclusive lock.

Parameters:
  • name (str) – The name for the lock file.

  • nonblocking (bool) – Return None instead of waiting indefinitely to get the exclusive lock if the lock is already taken.

Returns:

a file descriptor for a lock file after the file has been locked exclusively. In non-blocking mode None is returned if the lock cannot be gained.

Return type:

file or None

The returned file descriptor has to be kept. Otherwise the lock will be release automatically on file descriptor’s destruction.

>>> fd = get_lock('myapp')
>>> # ...... do some critical stuff ......
>>> release_lock(fd)
>>>
>>> fd = get_lock('myapp', nonblocking=True)
>>> if not fd:
>>>     print('cannot get lock')
>>> else:
>>>     # ...... do some critical stuff ......
>>>     release_lock(fd)
univention.lib.release_lock(fd: IO[str]) None[source]#

Releases the previously gained lock.

Parameters:

fd (file) – The file descriptor of the lock file.

Submodules#

univention.lib.account module#

python3 -m univention.lib.account lock –dn “$user_dn” –lock-time “$(date –utc ‘+%Y%m%d%H%M%SZ’)”

univention.lib.account.initialize_debug() None[source]#
univention.lib.account.lock(userdn: str, lock_timestamp: str) None[source]#

Lock a user account

  • used by ppolicy OpenLDAP overlay

  • used by PAM faillock

>>> from univention.lib.account import lock
>>> lock('uid=user1,dc=example,dc=com', '20141006192950Z')

univention.lib.admember module#

Python library for AD Member Mode.

univention.lib.admember.initialize_debug() None[source]#
exception univention.lib.admember.failedToSetService[source]#

Bases: Exception

ucs_addServiceToLocalhost failed

exception univention.lib.admember.invalidUCSServerRole[source]#

Bases: Exception

Invalid UCS Server Role

exception univention.lib.admember.failedADConnect[source]#

Bases: Exception

Connection to AD Server failed

exception univention.lib.admember.failedToSetAdministratorPassword[source]#

Bases: Exception

Failed to set the password of the UCS Administrator to the AD password

exception univention.lib.admember.failedToCreateAdministratorAccount[source]#

Bases: Exception

Failed to create the administrator account in UCS

exception univention.lib.admember.sambaSidNotSetForAdministratorAccount[source]#

Bases: Exception

sambaSID is not set for Administrator account in UCS

exception univention.lib.admember.failedToSearchForWellKnownSid[source]#

Bases: Exception

failed to search for well known SID

exception univention.lib.admember.failedToAddAdministratorAccountToDomainAdmins[source]#

Bases: Exception

failed to add Administrator account to Domain Admins

exception univention.lib.admember.domainnameMismatch[source]#

Bases: Exception

Domain Names don’t match

exception univention.lib.admember.connectionFailed[source]#

Bases: Exception

Connection to AD failed

exception univention.lib.admember.notDomainAdminInAD[source]#

Bases: Exception

User is not member of Domain Admins group in AD

exception univention.lib.admember.univentionSambaWrongVersion[source]#

Bases: Exception

univention-samba candidate has wrong version

exception univention.lib.admember.timeSyncronizationFailed[source]#

Bases: Exception

Time synchronization failed.

exception univention.lib.admember.manualTimeSyncronizationRequired[source]#

Bases: timeSyncronizationFailed

Time difference critical for Kerberos but synchronization aborted.

exception univention.lib.admember.sambaJoinScriptFailed[source]#

Bases: Exception

26univention-samba.inst failed

exception univention.lib.admember.failedToAddServiceRecordToAD[source]#

Bases: Exception

failed to add SRV record in AD

exception univention.lib.admember.failedToGetUcrVariable[source]#

Bases: Exception

failed to get ucr variable

univention.lib.admember.is_localhost_in_admember_mode(ucr: ConfigRegistry | None = None) bool[source]#
univention.lib.admember.is_localhost_in_adconnector_mode(ucr: ConfigRegistry | None = None) bool[source]#
univention.lib.admember.is_domain_in_admember_mode(ucr: ConfigRegistry | None = None) bool[source]#
univention.lib.admember.check_connection(ad_domain_info, username, password)[source]#
univention.lib.admember.flush_nscd_hosts_cache() None[source]#
univention.lib.admember.decode_sid(value)[source]#
univention.lib.admember.check_ad_account(ad_domain_info: dict[str, str], username: str, password: str, ucr: ConfigRegistry | None = None) bool[source]#

returns True if account is Administrator in AD returns False if account is just a member of Domain Admins raises exception notDomainAdminInAD if neither criterion is met.

univention.lib.admember.prepare_administrator(username: str, password: str, ucr: ConfigRegistry | None = None) None[source]#
univention.lib.admember.synchronize_account_position(ad_domain_info: dict[str, str], username: str, password: str, ucr: ConfigRegistry | None = None) bool[source]#
univention.lib.admember.server_supports_ssl(server: str) bool[source]#
univention.lib.admember.enable_ssl() None[source]#
univention.lib.admember.disable_ssl() None[source]#
univention.lib.admember.add_admember_service_to_localhost() None[source]#
univention.lib.admember.add_adconnector_service_to_localhost() None[source]#
univention.lib.admember.remove_admember_service_from_localhost() None[source]#
univention.lib.admember.info_handler(msg: Any) None[source]#
univention.lib.admember.error_handler(msg: Any) None[source]#
univention.lib.admember.remove_install_univention_samba(info_handler: Callable[..., None] = <function info_handler>, step_handler: Callable[..., None] | None = None, error_handler: Callable[..., None] = <function error_handler>, install: bool = True, uninstall: bool = True) bool[source]#
class univention.lib.admember.CLDAP_RES(forest, dns_domain, domain_name, pdc_dns_name, pdc_name, server_site, client_site)#

Bases: tuple

Create new instance of CLDAP_RES(forest, dns_domain, domain_name, pdc_dns_name, pdc_name, server_site, client_site)

client_site#

Alias for field number 6

dns_domain#

Alias for field number 1

domain_name#

Alias for field number 2

forest#

Alias for field number 0

pdc_dns_name#

Alias for field number 3

pdc_name#

Alias for field number 4

server_site#

Alias for field number 5

univention.lib.admember.cldap_finddc(ip: str) CLDAP_RES[source]#
univention.lib.admember.get_defaultNamingContext(ad_server_ip: str) str[source]#
univention.lib.admember.lookup_adds_dc(ad_server: str = '', ucr: ConfigRegistry | None = None, check_dns: bool = True) dict[str, str][source]#

CLDAP lookup

univention.lib.admember.set_timeserver(timeserver: str, ucr: ConfigRegistry | None = None) None[source]#
univention.lib.admember.stop_service(service: str) None[source]#
univention.lib.admember.start_service(service: str) None[source]#
univention.lib.admember.restart_service(service: str) None[source]#
univention.lib.admember.invoke_service(service: str, cmd: str) None[source]#
univention.lib.admember.do_time_sync(ad_ip: str) bool[source]#
univention.lib.admember.time_sync(ad_ip: str, tolerance: int = 180, critical_difference: int = 360) bool[source]#

Try to sync the local time with an AD server

univention.lib.admember.check_server_role(ucr: ConfigRegistry | None = None) None[source]#
univention.lib.admember.check_domain(ad_domain_info: dict[str, str], ucr: ConfigRegistry | None = None) None[source]#
univention.lib.admember.set_nameserver(server_ips: Iterable[str], ucr: ConfigRegistry | None = None) tuple[list[str], list[str]][source]#
univention.lib.admember.rename_well_known_sid_objects(username: str, password: str, ucr: ConfigRegistry | None = None) None[source]#
univention.lib.admember.make_deleted_objects_readable_for_this_machine(username: str, password: str, ucr: ConfigRegistry | None = None) None[source]#
univention.lib.admember.prepare_dns_reverse_settings(ad_domain_info: dict[str, str], ucr: ConfigRegistry | None = None) tuple[list[str], list[str]][source]#
univention.lib.admember.prepare_kerberos_ucr_settings(realm: str | None = None, ucr: ConfigRegistry | None = None) tuple[list[str], list[str]][source]#
univention.lib.admember.set_ucr(ucr_set: list[str], ucr_unset: list[str]) None[source]#
univention.lib.admember.prepare_ucr_settings() None[source]#
univention.lib.admember.revert_ucr_settings() None[source]#
univention.lib.admember.prepare_connector_settings(username: str, password: str, ad_domain_info: dict[str, str], ucr: ConfigRegistry | None = None) None[source]#
univention.lib.admember.revert_connector_settings(ucr: ConfigRegistry | None = None) None[source]#
univention.lib.admember.disable_local_samba4() None[source]#
univention.lib.admember.disable_local_heimdal() None[source]#
univention.lib.admember.run_samba_join_script(username: str, password: str, ucr: ConfigRegistry | None = None) None[source]#
univention.lib.admember.add_host_record_in_ad(uid: str | None = None, binddn: str | None = None, bindpw: str | None = None, bindpwdfile: str | None = None, fqdn: str | None = None, ip: str | None = None, sso: bool = False) bool[source]#
univention.lib.admember.get_domaincontroller_srv_record(domain: str, nameserver: str | None = None) bool | str | None[source]#
univention.lib.admember.add_domaincontroller_srv_record_in_ad(ad_ip: str, username: str, password: str, ucr: ConfigRegistry | None = None) bool[source]#
univention.lib.admember.get_ucr_variable_from_ucs(host: str, server: str, var: str) str[source]#
univention.lib.admember.set_nameserver_from_ucs_master(ucr: ConfigRegistry | None = None) None[source]#
univention.lib.admember.configure_ad_member(ad_server_ip: str, username: str, password: str) None[source]#
univention.lib.admember.configure_backup_as_ad_member() None[source]#
univention.lib.admember.configure_slave_as_ad_member() None[source]#
univention.lib.admember.configure_member_as_ad_member() None[source]#
univention.lib.admember.configure_container_as_ad_member() None[source]#
univention.lib.admember.revert_backup_ad_member() None[source]#
univention.lib.admember.revert_slave_ad_member() None[source]#
univention.lib.admember.revert_member_ad_member() None[source]#
univention.lib.admember.revert_container_ad_member() None[source]#

univention.lib.atjobs module#

Univention common Python library for handling at jobs.

This module abstracts the handling of at-jobs, each job is encapsulated by the class AtJob. Use the method add() in order to add a new command to the queue of at-jobs. Use the methods list() and load() to get a list of all registered jobs or to load a specific job given an ID, respectively. The module uses time stamps in seconds for scheduling jobs.

class univention.lib.atjobs.AtJob(nr: int, owner: str, execTime: datetime, isRunning: bool)[source]#

Bases: object

This class is an abstract representation of an at-job. Do not initiate the class directly, but use the methods provided in this module.

Parameters:
  • nr (int) – Job number.

  • owner (str) – User owning the job.

  • execTime (datetime.datetime) – Planned job execution time.

  • isRunning (bool) – True is the jub is currently running, False otherwise.

rm() int[source]#

Remove the job from the queue.

univention.lib.atjobs.add(cmd: str, execTime: float | datetime.datetime | None = None, comments: Mapping[str, str] | None = {}) AtJob | None[source]#

Add a new command to the job queue given a time at which the job will be executed.

Parameters:
  • execTime (int or float or datetime.datetime or None) – execution time either as seconds since the epoch or as a datetime.datetime instance. Defaults to now.

  • comments (dict) – A optional dictionary with comments to be associated with the job.

Returns:

The created job or None.

Return type:

AtJob or None

univention.lib.atjobs.list(extended: bool = False) builtins.list[AtJob][source]#

Returns a list of all registered jobs.

Parameters:

extended (bool) – If set to True also the comments and the command to execute are fetched.

Returns:

A list of AtJob instances.

Return type:

list[AtJob]

This can be used to re-schedule a job.

univention.lib.atjobs.load(nr: int, extended: bool = False) AtJob | None[source]#

Load the job given.

Parameters:
  • nr – Job number.

  • extended (bool) – If set to True also the comments and the command to execute are fetched.

Returns:

None if job does not exist, otherwise an instance of AtJob.

Return type:

AtJob

univention.lib.atjobs.remove(nr: int) int | None[source]#

Removes the at job with the given number.

Parameters:

nr (int) – Job number.

univention.lib.atjobs.reschedule(nr: int, execTime: float | None = None) AtJob | None[source]#

Re-schedules the at job with the given number for the specified time.

Parameters:
Returns:

The created job or None.

Return type:

AtJob or None

Raises:

AttributeError: if the job cannot be found.

univention.lib.fstab module#

Handle parsing and writing /etc/fstab.

See <http://linux.die.net/include/mntent.h>.

exception univention.lib.fstab.InvalidEntry[source]#

Bases: Exception

Invalid entry in file system table

class univention.lib.fstab.File(file: str = '/etc/fstab')[source]#

Bases: list

Handle lines of /etc/fstab.

Parameters:

file (str) – The name of the file.

load() None[source]#

Load entries from file.

find(**kargs: str) Entry | None[source]#

Search and return the entry matching the criteria.

Parameters:

kwargs – A mapping of Entry attributes to values.

Returns:

The first entry matching all criteria or None.

Return type:

Entry or None

get(filesystem: Container[str] = [], ignore_root: bool = True) list[Entry][source]#

Return list of entries matching a list of file system types.

Parameters:
  • filesystem (list[str]) – A list of file system names.

  • ignore_root (bool) – Skip the root file system if True.

Returns:

A list of matching entries.

Return type:

list[Entry]

save(filename: str | None = None) None[source]#

Save entries to file.

class univention.lib.fstab.Entry(spec: str, mount_point: str, fs_type: str, options: str | list = '', dump: str | None = None, passno: str | None = None, comment: str | None = None)[source]#

Bases: object

Mount table entry of fstab(5).

Parameters:
  • spec (str) – This field describes the block special device or remote filesystem to be mounted.

  • mount_point (str) – This field describes the mount point (target) for the filesystem.

  • type (str) – The type of the filesystem.

  • options (list[str]) – The list of mount options associated with the filesystem.

  • dump (int) – Option for dump(8).

  • passno (int) – Order information for fsck(8).

  • comment (str) – Optional comment from end of line.

Variables:

uuid (str) – The file system UUID if the file system is mounted by it. Otherwise None.

classmethod quote(s: str) str[source]#

Quote string to octal.

>>> Entry.quote('a b')
'a\\040b'
classmethod unquote(s: str) str[source]#

Unquote octal to string.

>>> Entry.unquote('a\040b')
'a b'
hasopt(opt: str) list[str][source]#

Search for an option matching OPT.

>>> Entry('/dev/sda', '/', 'ext3', 'default,ro,user_xattr,acl', 0, 0).hasopt('user')
['user_xattr']

univention.lib.i18n module#

Internationalization (i18n) utilities.

exception univention.lib.i18n.I18N_Error[source]#

Bases: Exception

Error in Internationalization.

class univention.lib.i18n.Locale(locale: str | None = None)[source]#

Bases: object

Represents a locale specification and provides simple access to language, territory, codeset and modifier.

Parameters:

locale (str or None) – The locale string language[_territory][.codeset][@modifier].

>>> Locale("deu_GER")
>>> str(Locale("ca_ES@valencia"))
>>> str(Locale(""))
REGEX = re.compile('^(?P<language>([a-z]{2}|C|POSIX))(?:_(?P<territory>[A-Z]{2}))?(?:\\.(?P<codeset>[a-zA-Z-0-9]+))?(?:@(?P<modifier>.+))?$')#
parse(locale: str) None[source]#

Parse locale string.

Parameters:

locale (str) – The locale string language[_territory][.codeset][@modifier].

Raises:
class univention.lib.i18n.NullTranslation(namespace: str, locale_spec: str | None = None, localedir: str | None = None)[source]#

Bases: object

Dummy translation.

Parameters:
  • namespace (str) – The name of the translation domain.

  • locale_spec (str) – The selected locale.

  • localedir (str) – The name of the directory containing the translation files.

property domain#
set_language(language: str = '') None[source]#

Select language.

Parameters:

language (str) – The language code.

property locale: Locale | None#

Return currently selected locale.

Returns:

The currently selected locale.

Return type:

Locale

translate(message: str) str[source]#

Translate message.

Parameters:

message (str) – The message to translate.

Returns:

The localized message.

Return type:

str

class univention.lib.i18n.Translation(*args, **kwargs)[source]#

Bases: NullTranslation

Translation.

locale: Locale = <univention.lib.i18n.Locale object>#
set_language(language: str = '') None[source]#

Select language.

Parameters:

language (str) – The language code.

Raises:

I18N_Error – if the given locale is not valid.

classmethod set_all_languages(language: str) None[source]#

Set the language of all existing Translation instances. This is required when instances are created during import time but later on the language should be changed.

univention.lib.ldap_extension module#

Python function to register UDM extensions in LDAP.

exception univention.lib.ldap_extension.BaseDirRestriction[source]#

Bases: Exception

univention.lib.ldap_extension.safe_path_join(basedir: str, filename: str) str[source]#
univention.lib.ldap_extension.set_handler_message(name: str, dn: str, msg: str) None[source]#
univention.lib.ldap_extension.get_handler_message(name: str, binddn: str, bindpw: str) dict[source]#
class univention.lib.ldap_extension.UniventionLDAPExtension(ucr: ConfigRegistry)[source]#

Bases: object

abstract property udm_module_name: str#
abstract property target_container_name: str#
abstract property active_flag_attribute: str#
abstract property filesuffix: str#
classmethod create_base_container(ucr: ConfigRegistry, udm_passthrough_options: list[str]) int[source]#
is_local_active() tuple[int, str | None][source]#
is_applicable_for_current_ucs_version(ucr: ConfigRegistry) bool[source]#
wait_for_activation(timeout: int = 180) bool[source]#
udm_find_object() tuple[int, str][source]#
udm_find_object_dn() tuple[int, str | None, str][source]#
ldap_touch_udm_object()[source]#
register(filename: str, options: Values, udm_passthrough_options: list[str], target_filename: str | None = None) None[source]#
unregister(objectname: str, options: Values, udm_passthrough_options: list[str]) None[source]#
mark_active(handler_name: str | None = None) None[source]#
class univention.lib.ldap_extension.UniventionLDAPExtensionWithListenerHandler(ucr: ConfigRegistry)[source]#

Bases: UniventionLDAPExtension

abstractmethod handler(dn: str, new: dict[str, list[bytes]], old: dict[str, list[bytes]], name: str = '') None[source]#
class univention.lib.ldap_extension.UniventionLDAPSchema(ucr: ConfigRegistry)[source]#

Bases: UniventionLDAPExtensionWithListenerHandler

target_container_name = 'ldapschema'#
udm_module_name = 'settings/ldapschema'#
active_flag_attribute = 'univentionLDAPSchemaActive'#
filesuffix = '.schema'#
basedir = '/var/lib/univention-ldap/local-schema'#
is_applicable_for_current_ucs_version(ucr: ConfigRegistry) bool[source]#
handler(dn: str, new: dict[str, list[bytes]], old: dict[str, list[bytes]], name: str = '') None[source]#
class univention.lib.ldap_extension.UniventionLDAPACL(ucr: ConfigRegistry)[source]#

Bases: UniventionLDAPExtensionWithListenerHandler

target_container_name = 'ldapacl'#
udm_module_name = 'settings/ldapacl'#
active_flag_attribute = 'univentionLDAPACLActive'#
filesuffix = '.acl'#
file_prefix = 'ldapacl_'#
handler(dn: str, new: dict[str, list[bytes]], old: dict[str, list[bytes]], name: str = '') None[source]#
class univention.lib.ldap_extension.UniventionDataExtension(ucr: ConfigRegistry)[source]#

Bases: UniventionLDAPExtension

target_container_name = 'data'#
udm_module_name = 'settings/data'#
active_flag_attribute = ''#
filesuffix = ''#
is_local_active() tuple[int, str | None][source]#

There is nothing to activate for a data extension, just pretend that everything is fine.

wait_for_activation(timeout: int = 180) bool[source]#
class univention.lib.ldap_extension.UniventionUDMExtension(ucr: ConfigRegistry)[source]#

Bases: UniventionLDAPExtension

target_subdir = ''#
property target_filepath: str#

return the most likely path where the listener will write the file to

wait_for_activation(timeout: int = 180) bool[source]#
class univention.lib.ldap_extension.UniventionUDMModule(ucr: ConfigRegistry)[source]#

Bases: UniventionUDMExtension

target_container_name = 'udm_module'#
udm_module_name = 'settings/udm_module'#
active_flag_attribute = 'univentionUDMModuleActive'#
filesuffix = '.py'#
target_udm_module = ''#
target_subdir = 'handlers'#
property target_filepath: str#

return the most likely path where the listener will write the file to

register(filename: str, options: Values, udm_passthrough_options: list[str], target_filename: str | None = None) None[source]#
wait_for_activation(timeout: int = 180) bool[source]#
class univention.lib.ldap_extension.UniventionUDMSyntax(ucr: ConfigRegistry)[source]#

Bases: UniventionUDMExtension

target_container_name = 'udm_syntax'#
udm_module_name = 'settings/udm_syntax'#
active_flag_attribute = 'univentionUDMSyntaxActive'#
filesuffix = '.py'#
target_subdir = 'syntax.d'#
class univention.lib.ldap_extension.UniventionUDMHook(ucr: ConfigRegistry)[source]#

Bases: UniventionUDMExtension

target_container_name = 'udm_hook'#
udm_module_name = 'settings/udm_hook'#
active_flag_attribute = 'univentionUDMHookActive'#
filesuffix = '.py'#
target_subdir = 'hooks.d'#
univention.lib.ldap_extension.option_validate_existing_filename(option: Option, opt: str, value: str) str[source]#
univention.lib.ldap_extension.option_validate_ucs_version(option: Option, opt: str, value: str) str[source]#
univention.lib.ldap_extension.option_validate_gnu_message_catalogfile(option: Option, opt: str, value: str) str[source]#
class univention.lib.ldap_extension.UCSOption(*opts, **attrs)[source]#

Bases: Option

TYPES = ('string', 'int', 'long', 'float', 'complex', 'choice', 'existing_filename', 'ucs_version')#
TYPE_CHECKER = {'choice': <function check_choice>, 'complex': <function check_builtin>, 'existing_filename': <function option_validate_existing_filename>, 'float': <function check_builtin>, 'gnu_message_catalogfile': <function option_validate_gnu_message_catalogfile>, 'int': <function check_builtin>, 'long': <function check_builtin>, 'ucs_version': <function option_validate_ucs_version>}#
univention.lib.ldap_extension.option_callback_udm_passthrough_options(option: Option, opt_str: str, value: str, parser: OptionParser, *args: list[str]) None[source]#
univention.lib.ldap_extension.check_data_module_options(option: Option, opt_str: str, value: str, parser: OptionParser) None[source]#
univention.lib.ldap_extension.option_callback_set_data_module_options(option: Option, opt_str: str, value: str, parser: OptionParser) None[source]#
univention.lib.ldap_extension.option_callback_append_data_module_options(option: Option, opt_str: str, value: str, parser: OptionParser) None[source]#
univention.lib.ldap_extension.check_udm_module_options(option: Option, opt_str: str, value: str, parser: OptionParser) None[source]#
univention.lib.ldap_extension.option_callback_set_udm_module_options(option: Option, opt_str: str, value: str, parser: OptionParser) None[source]#
univention.lib.ldap_extension.option_callback_append_udm_module_options(option: Option, opt_str: str, value: str, parser: OptionParser) None[source]#
univention.lib.ldap_extension.check_udm_syntax_options(option: Option, opt_str: str, value: str, parser: OptionParser) None[source]#
univention.lib.ldap_extension.option_callback_append_udm_syntax_options(option: Option, opt_str: str, value: str, parser: OptionParser) None[source]#
univention.lib.ldap_extension.check_udm_hook_options(option: Option, opt_str: str, value: str, parser: OptionParser) None[source]#
univention.lib.ldap_extension.option_callback_append_udm_hook_options(option: Option, opt_str: str, value: str, parser: OptionParser) None[source]#
univention.lib.ldap_extension.ucs_registerLDAPExtension() None[source]#
univention.lib.ldap_extension.ucs_unregisterLDAPExtension() None[source]#

univention.lib.license_tools module#

exception univention.lib.license_tools.LicenseCheckError[source]#

Bases: Exception

Generic error during license check

exception univention.lib.license_tools.LicenseExpired[source]#

Bases: LicenseCheckError

The license is expired

exception univention.lib.license_tools.LicenseNotFound[source]#

Bases: LicenseCheckError

The license cannot be found in LDAP

univention.lib.license_tools.is_CSP_license(lo: access | None = None) bool[source]#

Function to detect if installed license is a cloud service provider license (CSP).

Parameters:

lo (univention.uldap.access) – Optional LDAP connection to re-use. Otherwise a new LDAP connection with machine credentials is created.

Returns:

True if a valid CSP license has been found or False if a valid non-CSP license has been found.

Raises:

univention.lib.listenerSharePath module#

Univention Helper functions for creating or renaming share directories

univention.lib.listenerSharePath.dirIsMountPoint(path: str) str | None[source]#

Check if path is a mount point.

Parameters:

path (str) – The path to check.

Returns:

A string if the path is a mount point, None otherwise.

Return type:

str or None

univention.lib.listenerSharePath.checkDirFileSystem(path: str, cr: ConfigRegistry) str | None[source]#

Check if the given path is of a known file system type.

Parameters:
Returns:

A string if the path is a known file system, None otherwise.

Return type:

str or None

univention.lib.listenerSharePath.createOrRename(old: dict[str, list[bytes]], new: dict[str, list[bytes]], cr: ConfigRegistry) str | None[source]#

Create or rename a share.

Parameters:
Returns:

A string if an error occurs, None on success.

Return type:

str or None

univention.lib.listenerSharePath.is_blacklisted(path: str, ucr: ConfigRegistry) bool[source]#
>>> is_blacklisted('/home/', {})
True
>>> is_blacklisted('/home/', {'listener/shares/whitelist/defaults': '/home/*:/var/*'})
False
>>> is_blacklisted('/home', {})
True
>>> is_blacklisted('/home', {'listener/shares/whitelist/defaults': '/home/*:/var/*'})
False
>>> is_blacklisted('/home/Administrator', {})
True
>>> is_blacklisted('/home/Administrator', {'listener/shares/whitelist/defaults': '/home/*:/var/*'})
False
>>> is_blacklisted('/home/Administrator/', {'listener/shares/whitelist/admin': '/home/Administrator'})
False
>>> is_blacklisted('/var/', {'listener/shares/whitelist/univention-printserver-pdf': '/var/spool/cups-pdf/*'})
True
>>> is_blacklisted('/var', {'listener/shares/whitelist/univention-printserver-pdf': '/var/spool/cups-pdf/*'})
True
>>> is_blacklisted('/var/spool/', {'listener/shares/whitelist/univention-printserver-pdf': '/var/spool/cups-pdf/*'})
True
>>> is_blacklisted('/var/spool/cups-pdf/', {'listener/shares/whitelist/univention-printserver-pdf': '/var/spool/cups-pdf/*'})
False

univention.lib.locking module#

Univention Common Python Library for file locking

univention.lib.locking.get_lock(name: str, nonblocking: bool = False) IO[str] | None[source]#

Get a exclusive lock.

Parameters:
  • name (str) – The name for the lock file.

  • nonblocking (bool) – Return None instead of waiting indefinitely to get the exclusive lock if the lock is already taken.

Returns:

a file descriptor for a lock file after the file has been locked exclusively. In non-blocking mode None is returned if the lock cannot be gained.

Return type:

file or None

The returned file descriptor has to be kept. Otherwise the lock will be release automatically on file descriptor’s destruction.

>>> fd = get_lock('myapp')
>>> # ...... do some critical stuff ......
>>> release_lock(fd)
>>>
>>> fd = get_lock('myapp', nonblocking=True)
>>> if not fd:
>>>     print('cannot get lock')
>>> else:
>>>     # ...... do some critical stuff ......
>>>     release_lock(fd)
univention.lib.locking.release_lock(fd: IO[str]) None[source]#

Releases the previously gained lock.

Parameters:

fd (file) – The file descriptor of the lock file.

univention.lib.misc module#

Univention Common Python Library

univention.lib.misc.primaryVersionGreaterEqual(version: str) bool[source]#

Returns True if UCS_Version of primary is greater or equal to given version.

Parameters:

version (UCS_Version) – the UCS version to check

Returns:

True if UCS version of primary is greater or equal version

Return type:

bool

univention.lib.misc.createMachinePassword() str[source]#

Returns a $(pwgen) generated password according to the requirements in UCR variables machine/password/length and machine/password/complexity.

Returns:

A password.

Return type:

str

univention.lib.misc.getLDAPURIs(ucr: ConfigRegistry | None = None, sep: str = ' ') str[source]#

Returns a space separated list of all configured LDAP servers, according to UCR variables ldap/server/name and ldap/server/addition.

Parameters:

ucr (ConfigRegistry) – An optional UCR instance.

Returns:

A space separated list of LDAP URI.

Return type:

str

univention.lib.misc.getLDAPServersCommaList(ucr: ConfigRegistry | None = None) str[source]#

Returns a comma-separated string with all configured LDAP servers, ldap/server/name and ldap/server/addition.

Parameters:

ucr (ConfigRegistry) – An optional UCR instance.

Returns:

A space separated list of LDAP host names.

Return type:

str

univention.lib.misc.custom_username(name: str, ucr: ConfigRegistry | None = None) str[source]#

Returns the customized user name configured via UCR users/default/*.

Parameters:
Returns:

The translated user name.

Return type:

str

Raises:

ValueError – if no name is given.

univention.lib.misc.custom_groupname(name: str, ucr: ConfigRegistry | None = None) str[source]#

Returns the customized group name configured via UCR groups/default/*.

Parameters:
Returns:

The translated group name.

Return type:

str

Raises:

ValueError – if no name is given.

univention.lib.package_manager module#

Univention common Python Library for package management (info/install/progress…)

exception univention.lib.package_manager.LockError[source]#

Bases: Exception

Lock error for the package manager. Not to be confused with LockFailedException

class univention.lib.package_manager.PackageManager(lock: bool = True, info_handler: Callable[[...], None] | None = None, step_handler: Callable[[...], None] | None = None, error_handler: Callable[[...], None] | None = None, always_noninteractive: bool = True)[source]#

Bases: object

High-level package manager for UCS.

Parameters:
  • lock (bool) – Get an exclusive lock to prevent other instances from running in parallel.

  • info_handler – Some handler to handle info messages.

  • step_handler – Some handler to handle progress messages.

  • error_handler – Some handler to handle error messages.

  • always_noninteractive – Run dpkg in non-interactive mode to prevent any user input.

property cache: Cache#
always_install(pkgs: Iterable[Package] = (), just_mark: bool = False) None[source]#

Set packages that should be installed and never uninstalled. If you overwrite old always_install, make sure to call reopen_cache().

Parameters:
  • pkgs (list) – The list of packages to .

  • just_mark (bool) – if True, process the previously saved list of packages instead of the given new list.

lock(raise_on_fail: bool = True) bool[source]#

Get locks to prevent concurrent calls.

Parameters:

raise_on_fail (bool) – Raise LockError instead of returning False.

Returns:

True if all locks were are acquired, False otherwise.

Return type:

bool

Raises:

LockError – if the lock cannot be acquired.

unlock() bool[source]#

Release locks.

Returns:

True if the manager lock was taken, False otherwise.

Return type:

bool

is_locked() bool[source]#

Return the state of the lock.

Returns:

True if the lock is acquired, False otherwise.

Return type:

bool

locked(reset_status: bool = False, set_finished: bool = False) Iterator[None][source]#

Perform locking and cleanup actions before and after working with package state.

Parameters:
no_umc_restart(exclude_apache: bool = False) Iterator[None][source]#

Run package manager with restart of UMC (and Apache) disabled.

Parameters:

exclude_apache (bool) – DEPRECATED

add_hundred_percent() None[source]#

Add another 100 steps.

set_max_steps(steps: int) None[source]#

Set maximum number of steps.

Parameters:

steps (int) – Number of steps.

set_finished() None[source]#

Signal all steps done.

poll(timeout: Any) dict[str, Any][source]#

Poll for progress.

Parameters:

timeout – Ignored

Returns:

A dictionary containing the last info and error message, number of steps and finished state.

Return type:

dict

reset_status() None[source]#

Reset progress indicator back to start.

brutal_noninteractive() Iterator[None][source]#

Configure package manager to never ask for user input and to overwrite changed files

noninteractive() Iterator[None][source]#

Configure package manager to never ask for user input.

update() bool[source]#

apt-get update

Returns:

True on success, `False otherwise.

Return type:

bool

get_packages(pkg_names: Iterable[str | Package]) list[Package][source]#

Get many Package-objects at once (only those that exist, write error for others)

Parameters:

pkg_names – A list of binary package names.

Returns:

A list of APT cache entries.

get_package(pkg_name: str | Package, raise_key_error: bool = False) Package | None[source]#

Get Package-object for package name.

Parameters:
  • pkg_name (str) – A binary package name.

  • raise_key_error – Raise error when True, otherwise write an error message.

Returns:

The APT cache entry for the binary package.

is_installed(pkg_name: str | Package, reopen: bool = False) bool | None[source]#

Returns whether a package is installed.

Parameters:
  • pkg_name (str) – A binary package name.

  • reopen (bool) – Re-open the APT cache before checking.

Returns:

True if installed, False if not, and None if package is not found.

packages(reopen: bool = False) Iterator[Package][source]#

Yields all packages in cache.

Parameters:

reopen (bool) – Re-open the APT cache before returning.

mark_auto(auto: bool, *pkgs: str | Package) None[source]#

Immediately sets packages to automatically installed (or not).

Parameters:
  • auto (bool) – Mark the packages as automatically installed (True) or not.

  • pkgs – A list of binary package names.

Calls commit()!

mark(install: list[Package], remove: list[Package], dry_run: bool = False) tuple[list[str], list[str], list[str]][source]#

Mark packages as automatically installed (or not).

Parameters:
  • install – A list of packages to install.

  • remove – A list of packages to remove.

  • dry_run (bool) – Only simulate the action if True.

Returns:

A 3-tuple (to_be_installed, to_be_removed, broken), where each argument is a list of package names.

commit(install: Iterable[str | Package] = (), remove: Iterable[str | Package] = (), upgrade: bool = False, dist_upgrade: bool = False, msg_if_failed: str = '') bool[source]#

Really commit changes (mark_install or mark_delete) or pass Package-objects that shall be committed. Never forgets to pass progress objects, may print error messages, always reopens cache.

Parameters:
  • install – List of package names to install.

  • remove – List of package names to remove.

  • upgrade – Perform upgrade were no new packages are installed.

  • dist_upgrade – Perform upgrade were new packages may be installed and old packages may be removed.

  • msg_if_failed – Test message to output if things go wrong.

Returns:

True on success, False otherwise.

reopen_cache() None[source]#

Reopen the APT cache.

Has to be done when the APT database changed.

autoremove() bool[source]#

Remove all packages which are no longer required.

It seems that there is nothing like self.cache.autoremove.

upgrade() bool[source]#

Instantly performs an apt-get upgrade.

Returns:

True on success, False otherwise.

dist_upgrade() bool[source]#

Instantly performs an apt-get dist-upgrade.

Returns:

True on success, False otherwise.

install(*pkg_names: str | Package) bool[source]#

Instantly installs packages when found. Works like apt-get install and apt-get upgrade.

Parameters:

pkg_names – A list of binary package names to install.

Returns:

True on success, False otherwise.

uninstall(*pkg_names: str | Package) bool[source]#

Instantly deletes packages when found.

Parameters:

pkg_names – A list of binary package names to remove.

Returns:

True on success, False otherwise.

univention.lib.password module#

UDM library for changing user pasword

univention.lib.password.change(username: str, password: str) None[source]#

Change the password of the given user

>>> from univention.lib.password import change
>>> change('Administrator', 'secret12345')
>>> change('Administrator@DOMAIN.DE', 'secret12345')

univention.lib.policy_result module#

exception univention.lib.policy_result.PolicyResultFailed(message, returncode)[source]#

Bases: Exception

univention.lib.policy_result.policy_result(dn, binddn='', bindpw='', encoding='UTF-8', ldap_server=None)[source]#

Return a tuple of hash-lists, mapping attributes to a list of values and mapping attributes to the matching Policy-DN.

>>> (results, policies) = policy_result('dc=univention,dc=example')
>>> policies['univentionDhcpDomainNameServers']
'cn=default-settings,cn=dns,cn=dhcp,cn=policies,dc=univention,dc=example'
results['univentionDhcpDomainNameServers']
['192.168.0.111']
univention.lib.policy_result.ucr_policy_result(dn, binddn='', bindpw='', encoding='UTF-8', ldap_server=None)[source]#

Return a tuple of hash-lists, mapping attributes to a list of values and mapping attributes to the matching Policy-DN.

univention.lib.s4 module#

Univention common Python Library for common AD constants.

univention.lib.s4.well_known_sids = {'S-1-2': 'Local Authority', 'S-1-2-0': 'Local', 'S-1-2-1': 'Console Logon', 'S-1-3': 'Creator Authority', 'S-1-3-2': 'Creator Owner Server', 'S-1-3-3': 'Creator Group Server', 'S-1-4': 'Non-unique Authority', 'S-1-5': 'NT Authority', 'S-1-5-1': 'Dialup', 'S-1-5-32-544': 'Administrators', 'S-1-5-32-545': 'Users', 'S-1-5-32-546': 'Guests', 'S-1-5-32-547': 'Power Users', 'S-1-5-32-548': 'Account Operators', 'S-1-5-32-549': 'Server Operators', 'S-1-5-32-550': 'Print Operators', 'S-1-5-32-551': 'Backup Operators', 'S-1-5-32-552': 'Replicator', 'S-1-5-32-554': 'Pre-Windows 2000 Compatible Access', 'S-1-5-32-555': 'Remote Desktop Users', 'S-1-5-32-556': 'Network Configuration Operators', 'S-1-5-32-557': 'Incoming Forest Trust Builders', 'S-1-5-32-558': 'Performance Monitor Users', 'S-1-5-32-559': 'Performance Log Users', 'S-1-5-32-560': 'Windows Authorization Access Group', 'S-1-5-32-561': 'Terminal Server License Servers', 'S-1-5-32-562': 'Distributed COM Users', 'S-1-5-32-568': 'IIS_IUSRS', 'S-1-5-32-569': 'Cryptographic Operators', 'S-1-5-32-573': 'Event Log Readers', 'S-1-5-32-574': 'Certificate Service DCOM Access', 'S-1-5-80-0': 'All Services'}#

Well known security identifiers.

univention.lib.s4.well_known_domain_rids = {'498': 'Enterprise Read-only Domain Controllers', '500': 'Administrator', '501': 'Guest', '502': 'KRBTGT', '512': 'Domain Admins', '513': 'Domain Users', '514': 'Domain Guests', '515': 'Domain Computers', '516': 'Domain Controllers', '517': 'Cert Publishers', '518': 'Schema Admins', '519': 'Enterprise Admins', '520': 'Group Policy Creator Owners', '521': 'Read-Only Domain Controllers', '522': 'Cloneable Domain Controllers', '553': 'RAS and IAS Servers', '571': 'Allowed RODC Password Replication Group', '572': 'Denied RODC Password Replication Group'}#

Mapping of well known relative (security) identifiers to their (English) names.

See rids_for_well_known_security_identifiers for the reverse mapping.

univention.lib.s4.rids_for_well_known_security_identifiers = {'administrator': '500', 'allowed rodc password replication group': '571', 'cert publishers': '517', 'cloneable domain controllers': '522', 'denied rodc password replication group': '572', 'domain admins': '512', 'domain computers': '515', 'domain controllers': '516', 'domain guests': '514', 'domain users': '513', 'enterprise admins': '519', 'enterprise read-only domain controllers': '498', 'group policy creator owners': '520', 'guest': '501', 'krbtgt': '502', 'ras and ias servers': '553', 'read-only domain controllers': '521', 'schema admins': '518'}#

Mapping of lower cases English names to to well known relative (security) identifiers.

See well_known_domain_rids for the reverse mapping.

univention.lib.share_restrictions module#

class univention.lib.share_restrictions.Restrictions(name)[source]#

Bases: dict

INVALID_USERS = 'invalid users'#
VALID_USERS = 'valid users'#
HOSTS_DENY = 'hosts deny'#
HOSTS_ALLOW = 'hosts allow'#
property invalid_users#
property valid_users#
property hosts_deny#
property hosts_allow#
class univention.lib.share_restrictions.Share(name)[source]#

Bases: Restrictions

class univention.lib.share_restrictions.Printer(name)[source]#

Bases: Restrictions

property smbname#
class univention.lib.share_restrictions.ShareConfiguration[source]#

Bases: object

SHARES_DIR = '/etc/samba/local.config.d'#
SHARES_UDM_DIR = '/etc/samba/shares.conf.d'#
PRINTERS_UDM_DIR = '/etc/samba/printers.conf.d'#
POSTFIX = '.local.config.conf'#
PREFIX = 'printer.'#
INCLUDE_CONF = '/etc/samba/local.config.conf'#
GLOBAL_CONF = '/etc/samba/local.config.d/global.local.config.conf'#
CUPS_CONF = '/etc/cups/printers.conf'#
delete()[source]#

delete all conf’s in SHARES_DIR and INCLUDE_CONF

read_shares()[source]#

get invalid user from samba share conf

read_printers()[source]#

get invalid/valid users from cups and samba config

read_ucr()[source]#
read()[source]#
write()[source]#
property globals#
property shares#
property printers#

univention.lib.ucrLogrotate module#

Univention common Python library for logrotate configuration files.

univention.lib.ucrLogrotate.getLogrotateConfig(name: str, configRegistry: ConfigRegistry) dict[str, str][source]#

Build aggregated configuration for log file rotation.

Parameters:
  • name (str) – The name of the log file or service.

  • configRegistry (ConfigRegistry) – An UCR instance.

Returns:

A dictionary containing the merged configuration.

Return type:

dict

>>> ucr = ConfigRegistry()
>>> ucr.load()
>>> conf = getLogrotateConfig('service', ucr)

univention.lib.ucs module#

UCS release version.

class univention.lib.ucs.UCS_Version(version: tuple[int, int, int] | list[int] | str | Self)[source]#

Bases: object

Version object consisting of major-, minor-number and patch-level

Parameters:

version (list(int) or tuple(int) or str or UCS_Version) – must a str matching the pattern X.Y-Z or a triple with major, minor and patchlevel.

Raises:

TypeError – if the version cannot be parsed.

>>> v = UCS_Version((2,3,1))
>>> UCS_Version([2,3,1]) == v
True
>>> UCS_Version("2.3-1") == v
True
>>> UCS_Version(v) == v
True
FORMAT = '%(major)d.%(minor)d'#
FULLFORMAT = '%(major)d.%(minor)d-%(patchlevel)d'#
property mm: tuple[int, int]#

2-tuple (major, minor) version

property mmp: tuple[int, int, int]#

3-tuple (major, minor, patch-level) version

set(version: str) None[source]#

Parse string and set version.

Parameters:

version (str) – A UCS release version string.

Raises:

ValueError – if the string is not a valid UCS release version string.

univention.lib.umc module#

Univention common Python library to manage connections to remote UMC servers.

>>> umc = Client()
>>> umc.authenticate_with_machine_account()
>>> response = umc.umc_get('session-info')
>>> response.status
200
>>> response = umc.umc_logout()
>>> response.status
303
exception univention.lib.umc.ConnectionError(msg: str, reason: Exception | None = None)[source]#

Bases: Exception

Signal an error during connection setup.

Parameters:
  • msg (str) – A message string.

  • reason – The optional underlying exception.

exception univention.lib.umc.HTTPError(request, response, hostname)[source]#

Bases: Exception

Base class for HTTP errors. A specialized sub-class if automatically instantiated based on the HTTP return code.

Parameters:
  • request – The HTTP request.

  • response (http.client.HTTPResponse) – The HTTP response.

  • hostname (str) – The host name of the failed server.

codes: dict[int, type[Self]] = {300: <class 'univention.lib.umc.HTTPRedirect'>, 301: <class 'univention.lib.umc.MovedPermanently'>, 302: <class 'univention.lib.umc.Found'>, 303: <class 'univention.lib.umc.SeeOther'>, 304: <class 'univention.lib.umc.NotModified'>, 400: <class 'univention.lib.umc.BadRequest'>, 401: <class 'univention.lib.umc.Unauthorized'>, 403: <class 'univention.lib.umc.Forbidden'>, 404: <class 'univention.lib.umc.NotFound'>, 405: <class 'univention.lib.umc.MethodNotAllowed'>, 406: <class 'univention.lib.umc.NotAcceptable'>, 422: <class 'univention.lib.umc.UnprocessableEntity'>, 500: <class 'univention.lib.umc.InternalServerError'>, 502: <class 'univention.lib.umc.BadGateway'>, 503: <class 'univention.lib.umc.ServiceUnavailable'>}#

Specialized sub-classes for individual HTTP error codes.

property status: int#

Return the HTTP status code.

Returns:

the numerical status code.

Return type:

int

property message: str#

Return the HTTP status message.

Returns:

the textual status message.

Return type:

str

property result: str#

Return the HTTP result.

Returns:

the result data

Return type:

str

exception univention.lib.umc.HTTPRedirect(request, response, hostname)[source]#

Bases: HTTPError

http.client.MULTIPLE_CHOICES HTTP/1.1, RFC 2616, Section 10.3.1

code = 300#
exception univention.lib.umc.MovedPermanently(request, response, hostname)[source]#

Bases: HTTPRedirect

http.client.MOVED_PERMANENTLY HTTP/1.1, RFC 2616, Section 10.3.2

code = 301#
exception univention.lib.umc.Found(request, response, hostname)[source]#

Bases: HTTPRedirect

http.client.FOUND HTTP/1.1, RFC 2616, Section 10.3.3

code = 302#
exception univention.lib.umc.SeeOther(request, response, hostname)[source]#

Bases: HTTPRedirect

http.client.SEE_OTHER HTTP/1.1, RFC 2616, Section 10.3.4

code = 303#
exception univention.lib.umc.NotModified(request, response, hostname)[source]#

Bases: HTTPRedirect

http.client.NOT_MODIFIED HTTP/1.1, RFC 2616, Section 10.3.5

code = 304#
exception univention.lib.umc.BadRequest(request, response, hostname)[source]#

Bases: HTTPError

http.client.BAD_REQUEST HTTP/1.1, RFC 2616, Section 10.4.1

code = 400#
exception univention.lib.umc.Unauthorized(request, response, hostname)[source]#

Bases: HTTPError

http.client.UNAUTHORIZED HTTP/1.1, RFC 2616, Section 10.4.2

code = 401#
exception univention.lib.umc.Forbidden(request, response, hostname)[source]#

Bases: HTTPError

http.client.UNAUTHORIZED HTTP/1.1, RFC 2616, Section 10.4.4

code = 403#
exception univention.lib.umc.NotFound(request, response, hostname)[source]#

Bases: HTTPError

http.client.NOT_FOUND HTTP/1.1, RFC 2616, Section 10.4.5

code = 404#
exception univention.lib.umc.MethodNotAllowed(request, response, hostname)[source]#

Bases: HTTPError

http.client.METHOD_NOT_ALLOWED HTTP/1.1, RFC 2616, Section 10.4.6

code = 405#
exception univention.lib.umc.NotAcceptable(request, response, hostname)[source]#

Bases: HTTPError

http.client.NOT_ACCEPTABLE HTTP/1.1, RFC 2616, Section 10.4.7

code = 406#
exception univention.lib.umc.UnprocessableEntity(request, response, hostname)[source]#

Bases: HTTPError

http.client.UNPROCESSABLE_ENTITY WEBDAV, RFC 22518, Section 10.3

code = 422#
exception univention.lib.umc.InternalServerError(request, response, hostname)[source]#

Bases: HTTPError

http.client.INTERNAL_SERVER_ERROR HTTP/1.1, RFC 2616, Section 10.5.1

code = 500#
exception univention.lib.umc.BadGateway(request, response, hostname)[source]#

Bases: HTTPError

http.client.BAD_GATEWAY HTTP/1.1, RFC 2616, Section 10.5.3

code = 502#
exception univention.lib.umc.ServiceUnavailable(request, response, hostname)[source]#

Bases: HTTPError

http.client.SERVICE_UNAVAILABLE HTTP/1.1, RFC 2616, Section 10.5.4

code = 503#
class univention.lib.umc.Request(method: str, path: str, data: bytes | None = None, headers: dict[str, str] | None = None)[source]#

Bases: object

The HTTP request.

Parameters:
  • method (str) – GET / POST / PUT / DELETE

  • path (str) – the relative path to /univention/.

  • data (str) – either the raw request payload or some data which must be encoded by get_body()

  • headers (dict) – a mapping of HTTP headers

get_body() bytes | None[source]#

Return the request data.

Returns:

encodes data in JSON if Content-Type wants it

Return type:

bytes

class univention.lib.umc.Response(status: int, reason: str, body: bytes, headers: list[tuple[str, str]], _response: HTTPResponse)[source]#

Bases: object

The HTTP response.

Parameters:
  • status (int) – HTTP status code between 200 and 599.

  • reason (str) – string with the reason phrase e.g. ‘OK’

  • body (bytes) – the raw response body

  • headers (list) – the response headers as list of tuples

  • _response (http.client.HTTPResponse) – The original HTTP response.

property result: Any#

Return result from JSON data.

Returns:

The result.

property message: Any#

Return message from JSON data.

Returns:

The message.

get_header(name: str) str | None[source]#
get_header(name: str, default: _T = None) _T

Return original HTTP response header.

Parameters:
  • name (str) – HTTP respone header name, e.g. Content-Type.

  • default – Default value of the header is not set. Defaults to None.

Returns:

The header value or None.

Return type:

str or None

decode_body() bytes | dict[source]#

Decode HTTP response and return JSON data as dictionary.

Returns:

JSON data is returned as a dictionary, all other as raw.

Return type:

dict or str

class univention.lib.umc.Client(hostname: str | None = None, username: str | None = None, password: str | None = None, language: str | None = None, timeout: float | None = None, automatic_reauthentication: bool = False)[source]#

Bases: object

A client capable to speak with a UMC server.

Parameters:
  • hostname (str) – The name of the host to connect. Defaults to the FQDN of the localhost.

  • username (str) – A user name.

  • password (str) – The password of the user.

  • language (str) – The preferred language.

  • timeout (float) – Set the default timeout in seconds (float) for new connections.

  • automatic_reauthentication (bool) – Automatically re-authenticate and re-do requests if the authentication cookie expires.

ConnectionType#

alias of HTTPSConnection

authenticate(username: str, password: str) Response[source]#

Authenticate against the host and preserves the cookie. Has to be done only once (but keep in mind that the session probably expires after 10 minutes of inactivity)

Parameters:
  • username (str) – A user name.

  • password (str) – The password of the user.

reauthenticate() Response[source]#

Re-authenticate using the stored username and password.

set_basic_http_authentication(username: str, password: str) None[source]#

Setup authentication using HTTP Basic authentication.

Parameters:
  • username (str) – A user name.

  • password (str) – The password of the user.

authenticate_saml(username: str, password: str) None[source]#

Setup authentication using SAML.

Parameters:
  • username (str) – A user name.

  • password (str) – The password of the user.

Warning

not implemented.

authenticate_with_machine_account() None[source]#

Setup authentication using the machine account.

Raises:

ConnectionError – if /etc/machine.secret cannot be read.

umc_command(path: str, options: dict | None = None, flavor: str | None = None, headers: dict | None = None) Response[source]#

Perform generic UMC command.

Parameters:
  • path (str) – The URL path of the command after the command/ prefix.

  • options (dict) – The argument for the UMC command.

  • flavor (str) – Optional name of the UMC module flavor, e.g. users/user for UDM modules.

  • headers (dict) – Optional HTTP headers.

Returns:

The UMC response.

Return type:

univention.lib.umc.Response

umc_set(options: dict | None, headers: dict | None = None) Response[source]#

Perform UMC set command.

Parameters:
  • options (dict) – The argument for the UMC set command.

  • headers (dict) – Optional HTTP headers.

Returns:

The UMC response.

Return type:

univention.lib.umc.Response

umc_set_password(options: dict | None, headers: dict | None = None) Response[source]#

Perform UMC set/password command. Target UMC version need to be >= UCS 5.0-4.

Parameters:
  • options (dict) – The argument for the UMC set command.

  • headers (dict) – Optional HTTP headers.

Returns:

The UMC response.

Return type:

univention.lib.umc.Response

umc_get(path: str, options: dict | None = None, headers: dict | None = None) Response[source]#

Perform UMC get command.

Parameters:
  • path (str) – The URL path of the command after the get/ prefix.

  • options (dict) – The argument for the UMC get command.

  • headers (dict) – Optional HTTP headers.

Returns:

The UMC response.

Return type:

univention.lib.umc.Response

umc_upload() None[source]#

Perform UMC upload action.

Warning

not implemented.

umc_auth(username: str, password: str, **data: str) Response[source]#

Perform UMC authentication command.

Parameters:
  • username (str) – A user name.

  • password (str) – The password of the user.

  • data – Additional argument for the UMC auth command.

Returns:

The UMC response.

Return type:

univention.lib.umc.Response

umc_logout() Response[source]#

Perform UMC logout action.

Returns:

The UMC response.

Return type:

univention.lib.umc.Response

request(method: str, path: str, data: Any = None, headers: dict | None = None) Response[source]#

Send request to UMC server handling re-authentication.

Parameters:
  • method (str) – The HTTP method for the request.

  • path (str) – The URL of the request.

  • data – The message body.

  • headers (dict) – Optional HTTP headers.

:raises univention.lib.umc.Unauthorized: if the session expired and re-authentication was disabled.

Returns:

The UMC response.

Return type:

univention.lib.umc.Response

send(request: Request) Response[source]#

Low-level function to send request to UMC server.

Parameters:

request (Request) – A UMC request.

Returns:

The UMC response.

Return type:

univention.lib.umc.Response

Raises:

univention.lib.umc_module module#

Univention common Python library with helper functions for MIME type handling.

univention.lib.umc_module.get_mime_type(data: bytes) str[source]#

Guess MIME type of data.

Parameters:

data (bytes) – Some data.

Returns:

The MIME type string.

Return type:

str

univention.lib.umc_module.get_mime_description(data: bytes) str[source]#

Guess type of data silimar to file.

Parameters:

data (bytes) – Some data.

Returns:

A descriptive string.

Return type:

str

univention.lib.umc_module.compression_mime_type_of_buffer(data: bytes) tuple[str, Callable[[Any], bytes]][source]#

Guess MIME type of compressed data.

Parameters:

data (bytes) – Some compressed data.

Returns:

A 2-tuple (mime_type, uncompress_function).

Return type:

tuple[str, str]

Raises:

univention.admin.uexceptions.valueError – if the compression format is not recognized.

univention.lib.umc_module.uncompress_buffer(data: bytes) tuple[str | None, bytes][source]#

Return uncompressed data and its MIME type.

Parameters:

data (bytes) – Some compressed data.

Returns:

A 2-tuple (mime_type, uncompressed_data). On errors mime_type is None and uncompressed_data is data.

Return type:

tuple[str, bytes]

univention.lib.umc_module.uncompress_file(filename: str) tuple[str | None, bytes][source]#

Return uncompressed file content and its MIME type.

Parameters:

filename (str) – The name of the file.

Returns:

A 2-tuple (mime_type, uncompressed_data). On errors mime_type is None and uncompressed_data is data.

Return type:

tuple[str, bytes]

univention.lib.umc_module.image_mime_type_of_buffer(data: bytes) str[source]#

Guess MIME type of image.

Parameters:

data (bytes) – Some image data.

Returns:

The MIME type string.

Return type:

str

Raises:

univention.admin.uexceptions.valueError – if the image format is not supported.

univention.lib.umc_module.imagedimensions_of_buffer(data: bytes) tuple[int, int][source]#

Return image dimension of image.

Parameters:

data (bytes) – Some image data.

Returns:

A 2-tuple (width, height)

Return type:

tuple[int, int]

univention.lib.umc_module.imagecategory_of_buffer(data: bytes) tuple[str, str | None, str] | None[source]#

Return MIME types and size information for image.

Strparam bytes data:

Some (compressed) image data.

Returns:

a 3-tuple (image_mime_type, compression_mime_type, dimension) where dimension is {width}x{height} or scalable. None if the format is not recognized.

Return type:

tuple[str, str, str]

univention.lib.umc_module.default_filename_suffix_for_mime_type(mime_type: str, compression_mime_type: str) str | None[source]#

Return default file name suffix for image.

Parameters:
  • mime_type (str) – The MIME type of the image.

  • compression_mime_type (str) – The MIME type of the compression.

Returns:

A suffix string or None if the image format is not supported.

Rytpe:

str