Skip to main content

Expression Policies

Expression policies are perhaps the most flexible way to define specific implementations of flows and stages. With Expression polices, you can provide Python code to enforce custom checks and validation.

The passing of the policy is determined by the return value of the code.

To pass a policy, use:

return True

To fail a policy use:

return False

Example: Here's a simple policy that you could bind to a MFA validation stage if you want to specify that only certain users should be prompted for MFA validation:

if request.context["pending_user"].username == "marie":
   return True
return False

If the return is True that means Yes, use this stage (i.e. the user will get the MFA prompt). A return of False means remove this stage from the plan.

Available Functions​

ak_message(message: str)​

Add a message, visible by the end user. This can be used to show the reason why they were denied.

Example:

ak_message("Access denied")
return False

regex_match(value: Any, regex: str) -> bool​

Check if value matches Regular Expression regex.

Example:

return regex_match(request.user.username, '.*admin.*')

regex_replace(value: Any, regex: str, repl: str) -> str​

Replace anything matching regex within value with repl and return it.

Example:

user_email_local = regex_replace(request.user.email, '(.+)@.+', '')

list_flatten(value: list[Any] | Any) -> Optional[Any]​

Flatten a list by either returning its first element, None if the list is empty, or the passed in object if its not a list.

Example:

user = list_flatten(["foo"])
# user = "foo"

ak_call_policy(name: str, **kwargs) -> PolicyResult authentik 2021.12+​

Call another policy with the name name. Current request is passed to policy. Key-word arguments can be used to modify the request's context.

Example:

result = ak_call_policy("test-policy")
# result is a PolicyResult object, so you can access `.passing` and `.messages`.
# Starting with authentik 2023.4 you can also access `.raw_result`, which is the raw value returned from the called policy
# `result.passing` will always be a boolean if the policy is passing or not.
return result.passing

result = ak_call_policy("test-policy-2", foo="bar")
# Inside the `test-policy-2` you can then use `request.context["foo"]`
return result.passing

ak_is_group_member(user: User, **group_filters) -> bool​

Check if user is member of a group matching **group_filters.

Example:

return ak_is_group_member(request.user, name="test_group")

ak_user_by(**filters) -> Optional[User]​

Fetch a user matching **filters.

Returns "None" if no user was found, otherwise returns the User object.

Example:

other_user = ak_user_by(username="other_user")

ak_user_has_authenticator(user: User, device_type: Optional[str] = None) -> bool authentik 2022.9+​

Check if a user has any authenticator devices. Only fully validated devices are counted.

Optionally, you can filter a specific device type. The following options are valid:

  • totp
  • duo
  • static
  • webauthn

Example:

return ak_user_has_authenticator(request.user)

ak_create_event(action: str, **kwargs) -> None authentik 2022.9+​

Create a new event with the action set to action. Any additional key-word parameters will be saved in the event context. Additionally, context will be set to the context in which this function is called.

Before saving, any data-structure which are not representable in JSON are flattened, and credentials are removed.

The event is saved automatically

Example:

ak_create_event("my_custom_event", foo=request.user)

Comparing IP Addresses​

To compare IP Addresses or check if an IP Address is within a given subnet, you can use the functions ip_address('192.0.2.1') and ip_network('192.0.2.0/24'). With these objects you can do arithmetic operations.

You can also check if an IP Address is within a subnet by writing the following:

ip_address('192.0.2.1') in ip_network('192.0.2.0/24')
# evaluates to True

DNS resolution and reverse DNS lookups authentik 2023.3+​

To resolve a hostname to a list of IP addresses, use the functions resolve_dns(hostname) and resolve_dns(hostname, ip_version).

resolve_dns("google.com")  # return a list of all IPv4 and IPv6 addresses
resolve_dns("google.com", 4)  # return a list of only IP4 addresses
resolve_dns("google.com", 6)  # return a list of only IP6 addresses

You can also do reverse DNS lookups.

note

Reverse DNS lookups may not return the expected host if the IP address is part of a shared hosting environment. See: https://stackoverflow.com/a/19867936

To perform a reverse DNS lookup use reverse_dns("192.0.2.0"). If no DNS records are found the original IP address is returned.

info

DNS resolving results are cached in memory. The last 32 unique queries are cached for up to 3 minutes.

Variables​

  • ak_logger: structlog BoundLogger. See (structlog documentation)

    Example:

    ak_logger.debug("This is a test message")
    ak_logger.warning("This will be logged with a warning level")
    ak_logger.info("Passing structured data", request=request)

  • requests: requests Session object. See (request documentation)

  • request: A PolicyRequest object, which has the following properties:

    • request.user: The current user, against which the policy is applied. See User

      caution

      When a policy is executed in the context of a flow, this will be set to the user initiaing request, and will only be changed by a user_login stage. For that reason, using this value in authentication flow policies may not return the expected user. Use context['pending_user'] instead; User Identification and other stages update this value during flow execution.

      If the user is not authenticated, this will be set to a user called AnonymousUser, which is an instance of authentik.core.models.User (authentik uses django-guardian for per-object permissions, see).

    • request.http_request: The Django HTTP Request. See Django documentation.

    • request.obj: A Django Model instance. This is only set if the policy is ran against an object.

    • request.context: A dictionary with dynamic data. This depends on the origin of the execution.

  • geoip: GeoIP dictionary. The following fields are available:

    info

    For basic country matching, consider using a GeoIP policy.

    • continent: a two character continent code like NA (North America) or OC (Oceania).
    • country: the two character ISO 3166-1 alpha code for the country.
    • lat: the approximate latitude of the location associated with the IP address.
    • long: the approximate longitude of the location associated with the IP address.
    • city: the name of the city. May be empty.
    return context["geoip"]["continent"] == "EU"

  • asn: ASN dictionary. The following fields are available:

    info

    For basic ASN matching, consider using a GeoIP policy.

    • asn: the autonomous system number associated with the IP address.
    • as_org: the organization associated with the registered autonomous system number for the IP address.
    • network: the network associated with the record. In particular, this is the largest network where all of the fields except ip_address have the same value.
    return context["asn"]["asn"] == 64496

  • ak_is_sso_flow: Boolean which is true if request was initiated by authenticating through an external provider.

  • ak_client_ip: Client's IP Address or 255.255.255.255 if no IP Address could be extracted. Can be compared, for example

    return ak_client_ip in ip_network('10.0.0.0/24')
    # or
    return ak_client_ip.is_private

    See also Python documentation

Additionally, when the policy is executed from a flow, every variable from the flow's current context is accessible under the context object.

This includes the following:

  • context['flow_plan']: The actual flow plan itself, can be used to inject stages.

    • context['flow_plan'].context: The context of the currently active flow, which differs from the policy context. Some fields of flow plan context are passed to the root context, and updated from it, like 'prompt_data', but not every variable
    • context['flow_plan'].context['redirect']: The URL the user should be redirected to after the flow execution succeeds. (Optional)

  • context['prompt_data']: Data which has been saved from a prompt stage or an external source. (Optional)

  • context['application']: The application the user is in the process of authorizing. (Optional)

  • context['source']: The source the user is authenticating/enrolling with. (Optional)

  • context['pending_user']: The currently pending user, see User

  • context['is_restored']: Contains the flow token when the flow plan was restored from a link, for example the user clicked a link to a flow which was sent by an email stage. (Optional)

  • context['auth_method']: Authentication method (this value is set by password stages) (Optional)

    Depending on method, context['auth_method_args'] is also set.

    Can be any of:

    • password: Standard password login

    • auth_mfa: MFA login (this method is only set if no password was used)

      Sets context['auth_method_args'] to

      {
          "mfa_devices": [
              {
                  "pk": 1,
                  "app": "otp_static",
                  "name": "Static Token",
                  "model_name": "staticdevice"
              }
          ]
      }

    • auth_webauthn_pwl: Password-less WebAuthn login

    • jwt: OAuth Machine-to-machine login via external JWT

    • app_password: App password (token)

      Sets context['auth_method_args'] to

      {
          "token": {
              "pk": "f6d639aac81940f38dcfdc6e0fe2a786",
              "app": "authentik_core",
              "name": "test (expires=2021-08-23 15:45:54.725880+00:00)",
              "model_name": "token"
          }
      }

    • ldap: LDAP bind authentication

      Sets context['auth_method_args'] to

      {
          "source": {} // Information about the source used
      }