Skip to content

Okta

This module contains Okta integration steps.

koheesio.sso.okta.LoggerOktaTokenFilter #

LoggerOktaTokenFilter(
    okta_object: OktaAccessToken, name: str = "OktaToken"
)

Filter which hides token value from log.

Source code in src/koheesio/sso/okta.py 
42
43
44
def __init__(self, okta_object: OktaAccessToken, name: str = "OktaToken"):
    self.__okta_object = okta_object
    super().__init__(name=name)

filter #

filter(record: LogRecord) -> bool
Source code in src/koheesio/sso/okta.py 
46
47
48
49
50
51
52
def filter(self, record: LogRecord) -> bool:
    # noinspection PyUnresolvedReferences
    if token := self.__okta_object.output.token:
        token_value = token.get_secret_value()
        record.msg = record.msg.replace(token_value, "<SECRET_TOKEN>")

    return True

koheesio.sso.okta.Okta #

Base Okta class

client_id class-attribute instance-attribute #

client_id: str = Field(
    default=...,
    alias="okta_id",
    description="Okta account ID",
)

client_secret class-attribute instance-attribute #

client_secret: SecretStr = Field(
    default=...,
    alias="okta_secret",
    description="Okta account secret",
    repr=False,
)

data class-attribute instance-attribute #

data: Optional[Union[Dict[str, str], str]] = Field(
    default={"grant_type": "client_credentials"},
    description="Data to be sent along with the token request",
)

koheesio.sso.okta.OktaAccessToken #

OktaAccessToken(**kwargs)

Get Okta authorization token

Example: 

token = (
    OktaAccessToken(
        url="https://org.okta.com",
        client_id="client",
        client_secret=SecretStr("secret"),
        params={
            "p1": "foo",
            "p2": "bar",
        },
    )
    .execute()
    .token
)

Source code in src/koheesio/sso/okta.py 
82
83
84
85
86
def __init__(self, **kwargs):  # type: ignore[no-untyped-def]
    _logger = LoggingFactory.get_logger(name=self.__class__.__name__, inherit_from_koheesio=True)
    logger_filter = LoggerOktaTokenFilter(okta_object=self)
    _logger.addFilter(logger_filter)
    super().__init__(**kwargs)

Output #

Output class for OktaAccessToken.

token class-attribute instance-attribute #

token: Optional[SecretStr] = Field(
    default=None, description="Okta authentication token"
)

execute #

execute() -> None

Execute an HTTP Post call to Okta service and retrieve the access token.

Source code in src/koheesio/sso/okta.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
def execute(self) -> None:
    """
    Execute an HTTP Post call to Okta service and retrieve the access token.
    """
    HttpPostStep.execute(self)

    # noinspection PyUnresolvedReferences
    status_code = self.output.status_code
    # noinspection PyUnresolvedReferences
    raw_payload = self.output.raw_payload

    if status_code != 200:
        raise HTTPError(
            f"Request failed with '{status_code}' code. Payload: {raw_payload}",
            response=self.output.response_raw,
            request=None,
        )

    # noinspection PyUnresolvedReferences
    json_payload = self.output.json_payload

    if token := json_payload.get("access_token"):
        self.output.token = SecretStr(token)
    else:
        raise ValueError(f"No 'access_token' found in the Okta response: {json_payload}")