authentik.providers.proxy.models

  1"""authentik proxy models"""
  2
  3import string
  4from collections.abc import Iterable
  5from random import SystemRandom
  6from urllib.parse import urljoin
  7from uuid import uuid4
  8
  9from django.db import models
 10from django.templatetags.static import static
 11from django.utils.translation import gettext as _
 12from rest_framework.serializers import Serializer
 13
 14from authentik.core.models import ExpiringModel
 15from authentik.crypto.models import CertificateKeyPair
 16from authentik.lib.models import DomainlessURLValidator, InternallyManagedMixin
 17from authentik.outposts.models import OutpostModel
 18from authentik.providers.oauth2.models import (
 19    ClientType,
 20    GrantType,
 21    OAuth2Provider,
 22    RedirectURI,
 23    RedirectURIMatchingMode,
 24    ScopeMapping,
 25)
 26
 27SCOPE_AK_PROXY = "ak_proxy"
 28OUTPOST_CALLBACK_SIGNATURE = "X-authentik-auth-callback"
 29
 30
 31class ProxySession(InternallyManagedMixin, ExpiringModel):
 32    """Session storage for proxyv2 outposts using PostgreSQL"""
 33
 34    uuid = models.UUIDField(default=uuid4, primary_key=True)
 35    session_key = models.TextField(unique=True, db_index=True)
 36    user_id = models.UUIDField(null=True, blank=True, db_index=True)
 37
 38    session_data = models.JSONField(default=dict, blank=True)
 39
 40    class Meta:
 41        verbose_name = _("Proxy Session")
 42        verbose_name_plural = _("Proxy Sessions")
 43        indexes = [
 44            models.Index(fields=["user_id"]),
 45        ]
 46
 47    def __str__(self) -> str:
 48        return f"Session {self.session_key[:8]}..."
 49
 50
 51def get_cookie_secret():
 52    """Generate random 32-character string for cookie-secret"""
 53    return "".join(SystemRandom().choice(string.ascii_uppercase + string.digits) for _ in range(32))
 54
 55
 56def _get_callback_url(uri: str) -> list[RedirectURI]:
 57    return [
 58        RedirectURI(
 59            RedirectURIMatchingMode.STRICT,
 60            urljoin(uri, "outpost.goauthentik.io/callback") + f"?{OUTPOST_CALLBACK_SIGNATURE}=true",
 61        ),
 62        RedirectURI(RedirectURIMatchingMode.STRICT, uri + f"?{OUTPOST_CALLBACK_SIGNATURE}=true"),
 63    ]
 64
 65
 66class ProxyMode(models.TextChoices):
 67    """All modes a Proxy provider can operate in"""
 68
 69    PROXY = "proxy"
 70    FORWARD_SINGLE = "forward_single"
 71    FORWARD_DOMAIN = "forward_domain"
 72
 73
 74class ProxyProvider(OutpostModel, OAuth2Provider):
 75    """Protect applications that don't support any of the other
 76    Protocols by using a Reverse-Proxy."""
 77
 78    internal_host = models.TextField(
 79        validators=[DomainlessURLValidator(schemes=("http", "https"))],
 80        blank=True,
 81    )
 82    external_host = models.TextField(validators=[DomainlessURLValidator(schemes=("http", "https"))])
 83    internal_host_ssl_validation = models.BooleanField(
 84        default=True,
 85        help_text=_("Validate SSL Certificates of upstream servers"),
 86        verbose_name=_("Internal host SSL Validation"),
 87    )
 88    mode = models.TextField(
 89        default=ProxyMode.PROXY,
 90        choices=ProxyMode.choices,
 91        help_text=_(
 92            "Enable support for forwardAuth in traefik and nginx auth_request. Exclusive with "
 93            "internal_host."
 94        ),
 95    )
 96
 97    skip_path_regex = models.TextField(
 98        default="",
 99        blank=True,
100        help_text=_(
101            "Regular expressions for which authentication is not required. "
102            "Each new line is interpreted as a new Regular Expression."
103        ),
104    )
105
106    intercept_header_auth = models.BooleanField(
107        default=True,
108        help_text=_(
109            "When enabled, this provider will intercept the authorization header and authenticate "
110            "requests based on its value."
111        ),
112    )
113    basic_auth_enabled = models.BooleanField(
114        default=False,
115        verbose_name=_("Set HTTP-Basic Authentication"),
116        help_text=_(
117            "Set a custom HTTP-Basic Authentication header based on values from authentik."
118        ),
119    )
120    basic_auth_user_attribute = models.TextField(
121        blank=True,
122        verbose_name=_("HTTP-Basic Username Key"),
123        help_text=_(
124            "User/Group Attribute used for the user part of the HTTP-Basic Header. "
125            "If not set, the user's Email address is used."
126        ),
127    )
128    basic_auth_password_attribute = models.TextField(
129        blank=True,
130        verbose_name=_("HTTP-Basic Password Key"),
131        help_text=_("User/Group Attribute used for the password part of the HTTP-Basic Header."),
132    )
133
134    certificate = models.ForeignKey(
135        CertificateKeyPair,
136        on_delete=models.SET_NULL,
137        null=True,
138        blank=True,
139    )
140
141    cookie_secret = models.TextField(default=get_cookie_secret)
142    cookie_domain = models.TextField(default="", blank=True)
143
144    @property
145    def component(self) -> str:
146        return "ak-provider-proxy-form"
147
148    @property
149    def icon_url(self) -> str | None:
150        return static("authentik/sources/proxy.svg")
151
152    @property
153    def serializer(self) -> type[Serializer]:
154        from authentik.providers.proxy.api import ProxyProviderSerializer
155
156        return ProxyProviderSerializer
157
158    @property
159    def launch_url(self) -> str | None:
160        """Use external_host as launch URL"""
161        return self.external_host
162
163    def set_oauth_defaults(self):
164        """Ensure all OAuth2-related settings are correct"""
165        self.grant_types = [
166            GrantType.AUTHORIZATION_CODE,
167            GrantType.CLIENT_CREDENTIALS,
168            GrantType.PASSWORD,
169        ]
170        self.client_type = ClientType.CONFIDENTIAL
171        self.signing_key = None
172        self.include_claims_in_id_token = True
173        scopes = ScopeMapping.objects.filter(
174            managed__in=[
175                "goauthentik.io/providers/oauth2/scope-openid",
176                "goauthentik.io/providers/oauth2/scope-profile",
177                "goauthentik.io/providers/oauth2/scope-email",
178                "goauthentik.io/providers/oauth2/scope-entitlements",
179                "goauthentik.io/providers/proxy/scope-proxy",
180            ]
181        )
182        self.property_mappings.add(*list(scopes))
183        self.redirect_uris = _get_callback_url(self.external_host)
184
185    def __str__(self):
186        return f"Proxy Provider {self.name}"
187
188    def get_required_objects(self) -> Iterable[models.Model | str | tuple[str, models.Model]]:
189        required = [self]
190        if self.certificate is not None:
191            required.append(("authentik_crypto.view_certificatekeypair", self.certificate))
192            required.append(
193                ("authentik_crypto.view_certificatekeypair_certificate", self.certificate)
194            )
195            required.append(("authentik_crypto.view_certificatekeypair_key", self.certificate))
196        return required
197
198    class Meta:
199        verbose_name = _("Proxy Provider")
200        verbose_name_plural = _("Proxy Providers")
201        authentik_used_by_shadows = ["authentik_providers_oauth2.oauth2provider"]