authentik.lib.utils.http

http helpers

 1"""http helpers"""
 2
 3from uuid import uuid4
 4
 5from requests.sessions import PreparedRequest, Session
 6from structlog.stdlib import get_logger
 7
 8from authentik import authentik_full_version
 9from authentik.lib.config import CONFIG
10
11LOGGER = get_logger()
12
13
14def authentik_user_agent() -> str:
15    """Get a common user agent"""
16    return f"authentik@{authentik_full_version()}"
17
18
19class TimeoutSession(Session):
20    """Always set a default HTTP request timeout"""
21
22    def __init__(self, default_timeout=None):
23        super().__init__()
24        self.timeout = default_timeout
25
26    def send(
27        self,
28        request,
29        *,
30        stream=...,
31        verify=...,
32        proxies=...,
33        cert=...,
34        timeout=...,
35        allow_redirects=...,
36        **kwargs,
37    ):
38        if not timeout and self.timeout:
39            timeout = self.timeout
40        return super().send(
41            request,
42            stream=stream,
43            verify=verify,
44            proxies=proxies,
45            cert=cert,
46            timeout=timeout,
47            allow_redirects=allow_redirects,
48            **kwargs,
49        )
50
51
52class DebugSession(TimeoutSession):
53    """requests session which logs http requests and responses"""
54
55    def send(self, req: PreparedRequest, *args, **kwargs):
56        request_id = str(uuid4())
57        LOGGER.debug(
58            "HTTP request sent",
59            uid=request_id,
60            url=req.url,
61            method=req.method,
62            headers=req.headers,
63            body=req.body,
64        )
65        resp = super().send(req, *args, **kwargs)
66        LOGGER.debug(
67            "HTTP response received",
68            uid=request_id,
69            status=resp.status_code,
70            body=resp.text[: 32 * 1024],
71            headers=resp.headers,
72        )
73        return resp
74
75
76def get_http_session() -> Session:
77    """Get a requests session with common headers"""
78    session = TimeoutSession()
79    if CONFIG.get_bool("debug") or CONFIG.get("log_level") == "trace":
80        session = DebugSession()
81    session.headers["User-Agent"] = authentik_user_agent()
82    session.timeout = CONFIG.get_optional_int("http_timeout")
83    return session
LOGGER = <BoundLoggerLazyProxy(logger=None, wrapper_class=None, processors=None, context_class=None, initial_values={}, logger_factory_args=())>
def authentik_user_agent() -> str: 
15def authentik_user_agent() -> str:
16    """Get a common user agent"""
17    return f"authentik@{authentik_full_version()}"

Get a common user agent

class TimeoutSession(requests.sessions.Session): 
20class TimeoutSession(Session):
21    """Always set a default HTTP request timeout"""
22
23    def __init__(self, default_timeout=None):
24        super().__init__()
25        self.timeout = default_timeout
26
27    def send(
28        self,
29        request,
30        *,
31        stream=...,
32        verify=...,
33        proxies=...,
34        cert=...,
35        timeout=...,
36        allow_redirects=...,
37        **kwargs,
38    ):
39        if not timeout and self.timeout:
40            timeout = self.timeout
41        return super().send(
42            request,
43            stream=stream,
44            verify=verify,
45            proxies=proxies,
46            cert=cert,
47            timeout=timeout,
48            allow_redirects=allow_redirects,
49            **kwargs,
50        )

Always set a default HTTP request timeout

TimeoutSession(default_timeout=None) 
23    def __init__(self, default_timeout=None):
24        super().__init__()
25        self.timeout = default_timeout
timeout
def send( self, request, *, stream=Ellipsis, verify=Ellipsis, proxies=Ellipsis, cert=Ellipsis, timeout=Ellipsis, allow_redirects=Ellipsis, **kwargs): 
27    def send(
28        self,
29        request,
30        *,
31        stream=...,
32        verify=...,
33        proxies=...,
34        cert=...,
35        timeout=...,
36        allow_redirects=...,
37        **kwargs,
38    ):
39        if not timeout and self.timeout:
40            timeout = self.timeout
41        return super().send(
42            request,
43            stream=stream,
44            verify=verify,
45            proxies=proxies,
46            cert=cert,
47            timeout=timeout,
48            allow_redirects=allow_redirects,
49            **kwargs,
50        )

Send a given PreparedRequest.

:rtype: requests.Response

class DebugSession(TimeoutSession): 
53class DebugSession(TimeoutSession):
54    """requests session which logs http requests and responses"""
55
56    def send(self, req: PreparedRequest, *args, **kwargs):
57        request_id = str(uuid4())
58        LOGGER.debug(
59            "HTTP request sent",
60            uid=request_id,
61            url=req.url,
62            method=req.method,
63            headers=req.headers,
64            body=req.body,
65        )
66        resp = super().send(req, *args, **kwargs)
67        LOGGER.debug(
68            "HTTP response received",
69            uid=request_id,
70            status=resp.status_code,
71            body=resp.text[: 32 * 1024],
72            headers=resp.headers,
73        )
74        return resp

requests session which logs http requests and responses

def send(self, req: requests.models.PreparedRequest, *args, **kwargs): 
56    def send(self, req: PreparedRequest, *args, **kwargs):
57        request_id = str(uuid4())
58        LOGGER.debug(
59            "HTTP request sent",
60            uid=request_id,
61            url=req.url,
62            method=req.method,
63            headers=req.headers,
64            body=req.body,
65        )
66        resp = super().send(req, *args, **kwargs)
67        LOGGER.debug(
68            "HTTP response received",
69            uid=request_id,
70            status=resp.status_code,
71            body=resp.text[: 32 * 1024],
72            headers=resp.headers,
73        )
74        return resp

Send a given PreparedRequest.

:rtype: requests.Response

Inherited Members
TimeoutSession
TimeoutSession
timeout
def get_http_session() -> requests.sessions.Session: 
77def get_http_session() -> Session:
78    """Get a requests session with common headers"""
79    session = TimeoutSession()
80    if CONFIG.get_bool("debug") or CONFIG.get("log_level") == "trace":
81        session = DebugSession()
82    session.headers["User-Agent"] = authentik_user_agent()
83    session.timeout = CONFIG.get_optional_int("http_timeout")
84    return session

Get a requests session with common headers