Validate Ngrok authentication token

Validate rate limiting format

Initialize email sender with configuration. Args: config: EmailConfig object Raises: ValueError: If

Apply the selected theme and save preferences

Build a Permissions protobuf for share/update — single source of truth.

Resolve (password, used_existing) per cp-rotation-skip-feat: existing_password → rotation.password_complexity → DEFAULT_COMPLEXITY.

Scan running Docker containers for one with Guacamole DB env vars. Returns the container name, or raises CommandError if none found.

Dispatch a GatewayAction to the gateway and return the response.

Generic method to display results in table or JSON format.

Send an empty sync to the given PamEndpoints; expect a clean 200 (proto).

Execute a function with consistent error handling

Extract database-specific settings from pamSettings.connection (record JSON).

In headless/service mode, persistent login often needs a follow-up prompt (password, SSO, 2FA, device approval) that cannot be answered. Log once

Get the record key for a PAM config record.

Load configuration from YAML file.

Create a mock TypedRecord with an optional identityProviderUid custom field.

Create a real vault.TypedField.

Check that *secret_tags* (a plain dict) satisfies every (key, value) pair in *required_tags*. Azure and GCP both return tags/labels a

Point tunnel_registry_dir at tmp for the duration of a test.

Wrap _post_request_to_router with progressive backoff on 429 / throttle errors. Non-throttle errors propagate immediately. Final retry's exception

Rewrite folder_path roots to match actual shared folder names. Needed when edit.py adds a #N dedup suffix to the project folder, maki

Run execute() with Azure internals mocked. *secret_meta*: list of {'name', 'tags'} dicts. *value_map*: name → secret value

Strip C0/C1 control characters (incl. ESC) from attacker-supplied text.

True when AD context is set AND existing_password is NOT — gates the rm-create-user push-to-target route. See SECURITY_ARCHITECTURE.md.

True when rotation: present and rotate_on_provision is not False.

Print interactive prompt text to stderr so it's always visible and never pollutes stdout.

Stop the TOTP auto-refresh timer

Validate email section and check for email config existence.

Validate rotation section (schedule/on_demand, complexity, flags).

Clean up temporary files. Args: temp_files: List of temporary file paths to clean up

Convert conversationId to messageId format (replace + with -, / with _)

Create legacy blueprint for synchronous command execution.

Encode a Guacamole instruction to send to the server. Args: opcode: Instruction opcode *args: Instruction ar

Extract and sort devices from API response.

Format timeout as instance of timedelta for output to console timeout_delta(timedelta): Timeout setting as instance of timedelta. Returns str

Get configuration format based on existing files.

Get PAM configuration records. The default it to find all the record version 6 records. If the environment variable `PAM_REC

Find the highest DAGEdge version of all edge types. :param head_uid: :return:

Get a record dict from nested_share_records or record_cache.

True when settings are absent or match the vault empty/default template.

Parse expiration time string into timedelta

Read command with support for line continuation using backslash.

Read configuration parameter from service config file.

Reads an item from a stream and returns it as a string of bytes.

Return server-assigned rotation revision from cache, syncing when missing (not sequential).