Module aws_lambda_powertools.event_handler.openapi.swagger_ui.oauth2
Expand source code
# ruff: noqa: E501
import warnings
from typing import Dict, Optional, Sequence
from pydantic import BaseModel, Field, validator
from aws_lambda_powertools.event_handler.openapi.pydantic_loader import PYDANTIC_V2
from aws_lambda_powertools.shared.functions import powertools_dev_is_set
# Based on https://swagger.io/docs/open-source-tools/swagger-ui/usage/oauth2/
class OAuth2Config(BaseModel):
"""
OAuth2 configuration for Swagger UI
"""
# The client ID for the OAuth2 application
clientId: Optional[str] = Field(alias="client_id", default=None)
# The client secret for the OAuth2 application. This is sensitive information and requires the explicit presence
# of the POWERTOOLS_DEV environment variable.
clientSecret: Optional[str] = Field(alias="client_secret", default=None)
# The realm in which the OAuth2 application is registered. Optional.
realm: Optional[str] = Field(default=None)
# The name of the OAuth2 application
appName: str = Field(alias="app_name")
# The scopes that the OAuth2 application requires. Defaults to an empty list.
scopes: Sequence[str] = Field(default=[])
# Additional query string parameters to be included in the OAuth2 request. Defaults to an empty dictionary.
additionalQueryStringParams: Dict[str, str] = Field(alias="additional_query_string_params", default={})
# Whether to use basic authentication with the access code grant type. Defaults to False.
useBasicAuthenticationWithAccessCodeGrant: bool = Field(
alias="use_basic_authentication_with_access_code_grant",
default=False,
)
# Whether to use PKCE with the authorization code grant type. Defaults to False.
usePkceWithAuthorizationCodeGrant: bool = Field(alias="use_pkce_with_authorization_code_grant", default=False)
if PYDANTIC_V2:
model_config = {"extra": "allow"}
else:
class Config:
extra = "allow"
allow_population_by_field_name = True
@validator("clientSecret", always=True)
def client_secret_only_on_dev(cls, v: Optional[str]) -> Optional[str]:
if not v:
return None
if not powertools_dev_is_set():
raise ValueError(
"cannot use client_secret without POWERTOOLS_DEV mode. See "
"https://docs.powertools.aws.dev/lambda/python/latest/#optimizing-for-non-production-environments",
)
else:
warnings.warn(
"OAuth2Config is using client_secret and POWERTOOLS_DEV is set. This reveals sensitive information. "
"DO NOT USE THIS OUTSIDE LOCAL DEVELOPMENT",
stacklevel=2,
)
return v
def generate_oauth2_redirect_html() -> str:
"""
Generates the HTML content for the OAuth2 redirect page.
Source: https://github.com/swagger-api/swagger-ui/blob/master/dist/oauth2-redirect.html
"""
return """
<!doctype html>
<html lang="en-US">
<head>
<title>Swagger UI: OAuth2 Redirect</title>
</head>
<body>
<script>
'use strict';
function run () {
var oauth2 = window.opener.swaggerUIRedirectOauth2;
var sentState = oauth2.state;
var redirectUrl = oauth2.redirectUrl;
var isValid, qp, arr;
if (/code|token|error/.test(window.location.hash)) {
qp = window.location.hash.substring(1).replace('?', '&');
} else {
qp = location.search.substring(1);
}
arr = qp.split("&");
arr.forEach(function (v,i,_arr) { _arr[i] = '"' + v.replace('=', '":"') + '"';});
qp = qp ? JSON.parse('{' + arr.join() + '}',
function (key, value) {
return key === "" ? value : decodeURIComponent(value);
}
) : {};
isValid = qp.state === sentState;
if ((
oauth2.auth.schema.get("flow") === "accessCode" ||
oauth2.auth.schema.get("flow") === "authorizationCode" ||
oauth2.auth.schema.get("flow") === "authorization_code"
) && !oauth2.auth.code) {
if (!isValid) {
oauth2.errCb({
authId: oauth2.auth.name,
source: "auth",
level: "warning",
message: "Authorization may be unsafe, passed state was changed in server. The passed state wasn't returned from auth server."
});
}
if (qp.code) {
delete oauth2.state;
oauth2.auth.code = qp.code;
oauth2.callback({auth: oauth2.auth, redirectUrl: redirectUrl});
} else {
let oauthErrorMsg;
if (qp.error) {
oauthErrorMsg = "["+qp.error+"]: " +
(qp.error_description ? qp.error_description+ ". " : "no accessCode received from the server. ") +
(qp.error_uri ? "More info: "+qp.error_uri : "");
}
oauth2.errCb({
authId: oauth2.auth.name,
source: "auth",
level: "error",
message: oauthErrorMsg || "[Authorization failed]: no accessCode received from the server."
});
}
} else {
oauth2.callback({auth: oauth2.auth, token: qp, isValid: isValid, redirectUrl: redirectUrl});
}
window.close();
}
if (document.readyState !== 'loading') {
run();
} else {
document.addEventListener('DOMContentLoaded', function () {
run();
});
}
</script>
</body>
</html>
""".strip()
Functions
def generate_oauth2_redirect_html() ‑> str
-
Generates the HTML content for the OAuth2 redirect page.
Source: https://github.com/swagger-api/swagger-ui/blob/master/dist/oauth2-redirect.html
Expand source code
def generate_oauth2_redirect_html() -> str: """ Generates the HTML content for the OAuth2 redirect page. Source: https://github.com/swagger-api/swagger-ui/blob/master/dist/oauth2-redirect.html """ return """ <!doctype html> <html lang="en-US"> <head> <title>Swagger UI: OAuth2 Redirect</title> </head> <body> <script> 'use strict'; function run () { var oauth2 = window.opener.swaggerUIRedirectOauth2; var sentState = oauth2.state; var redirectUrl = oauth2.redirectUrl; var isValid, qp, arr; if (/code|token|error/.test(window.location.hash)) { qp = window.location.hash.substring(1).replace('?', '&'); } else { qp = location.search.substring(1); } arr = qp.split("&"); arr.forEach(function (v,i,_arr) { _arr[i] = '"' + v.replace('=', '":"') + '"';}); qp = qp ? JSON.parse('{' + arr.join() + '}', function (key, value) { return key === "" ? value : decodeURIComponent(value); } ) : {}; isValid = qp.state === sentState; if (( oauth2.auth.schema.get("flow") === "accessCode" || oauth2.auth.schema.get("flow") === "authorizationCode" || oauth2.auth.schema.get("flow") === "authorization_code" ) && !oauth2.auth.code) { if (!isValid) { oauth2.errCb({ authId: oauth2.auth.name, source: "auth", level: "warning", message: "Authorization may be unsafe, passed state was changed in server. The passed state wasn't returned from auth server." }); } if (qp.code) { delete oauth2.state; oauth2.auth.code = qp.code; oauth2.callback({auth: oauth2.auth, redirectUrl: redirectUrl}); } else { let oauthErrorMsg; if (qp.error) { oauthErrorMsg = "["+qp.error+"]: " + (qp.error_description ? qp.error_description+ ". " : "no accessCode received from the server. ") + (qp.error_uri ? "More info: "+qp.error_uri : ""); } oauth2.errCb({ authId: oauth2.auth.name, source: "auth", level: "error", message: oauthErrorMsg || "[Authorization failed]: no accessCode received from the server." }); } } else { oauth2.callback({auth: oauth2.auth, token: qp, isValid: isValid, redirectUrl: redirectUrl}); } window.close(); } if (document.readyState !== 'loading') { run(); } else { document.addEventListener('DOMContentLoaded', function () { run(); }); } </script> </body> </html> """.strip()
Classes
class OAuth2Config (**data: Any)
-
OAuth2 configuration for Swagger UI
Create a new model by parsing and validating input data from keyword arguments.
Raises ValidationError if the input data cannot be parsed to form a valid model.
Expand source code
class OAuth2Config(BaseModel): """ OAuth2 configuration for Swagger UI """ # The client ID for the OAuth2 application clientId: Optional[str] = Field(alias="client_id", default=None) # The client secret for the OAuth2 application. This is sensitive information and requires the explicit presence # of the POWERTOOLS_DEV environment variable. clientSecret: Optional[str] = Field(alias="client_secret", default=None) # The realm in which the OAuth2 application is registered. Optional. realm: Optional[str] = Field(default=None) # The name of the OAuth2 application appName: str = Field(alias="app_name") # The scopes that the OAuth2 application requires. Defaults to an empty list. scopes: Sequence[str] = Field(default=[]) # Additional query string parameters to be included in the OAuth2 request. Defaults to an empty dictionary. additionalQueryStringParams: Dict[str, str] = Field(alias="additional_query_string_params", default={}) # Whether to use basic authentication with the access code grant type. Defaults to False. useBasicAuthenticationWithAccessCodeGrant: bool = Field( alias="use_basic_authentication_with_access_code_grant", default=False, ) # Whether to use PKCE with the authorization code grant type. Defaults to False. usePkceWithAuthorizationCodeGrant: bool = Field(alias="use_pkce_with_authorization_code_grant", default=False) if PYDANTIC_V2: model_config = {"extra": "allow"} else: class Config: extra = "allow" allow_population_by_field_name = True @validator("clientSecret", always=True) def client_secret_only_on_dev(cls, v: Optional[str]) -> Optional[str]: if not v: return None if not powertools_dev_is_set(): raise ValueError( "cannot use client_secret without POWERTOOLS_DEV mode. See " "https://docs.powertools.aws.dev/lambda/python/latest/#optimizing-for-non-production-environments", ) else: warnings.warn( "OAuth2Config is using client_secret and POWERTOOLS_DEV is set. This reveals sensitive information. " "DO NOT USE THIS OUTSIDE LOCAL DEVELOPMENT", stacklevel=2, ) return v
Ancestors
- pydantic.main.BaseModel
- pydantic.utils.Representation
Class variables
var Config
var additionalQueryStringParams : Dict[str, str]
var appName : str
var clientId : Optional[str]
var clientSecret : Optional[str]
var realm : Optional[str]
var scopes : Sequence[str]
var useBasicAuthenticationWithAccessCodeGrant : bool
var usePkceWithAuthorizationCodeGrant : bool
Static methods
def client_secret_only_on_dev(v: Optional[str]) ‑> Optional[str]
-
Expand source code
@validator("clientSecret", always=True) def client_secret_only_on_dev(cls, v: Optional[str]) -> Optional[str]: if not v: return None if not powertools_dev_is_set(): raise ValueError( "cannot use client_secret without POWERTOOLS_DEV mode. See " "https://docs.powertools.aws.dev/lambda/python/latest/#optimizing-for-non-production-environments", ) else: warnings.warn( "OAuth2Config is using client_secret and POWERTOOLS_DEV is set. This reveals sensitive information. " "DO NOT USE THIS OUTSIDE LOCAL DEVELOPMENT", stacklevel=2, ) return v