Public Member Functions
def	do_GET (self)

str	client_id (self)

def	do_POST (self)

JsonObject	config (self)

JsonObject	authorization (self)

JsonObject	token (self)

Data Fields
	path

Static Public Attributes
	JsonObject = dict[str, object]

Private Member Functions
def	_check_issuer (self)

def	_check_authn (self)

dict[str, str]	_parse_params (self)

bool	_should_modify (self)

def	_get_param (self, name, default)

str	_content_type (self)

int	_interval (self)

str	_retry_code (self)

str	_uri_spelling (self)

def	_response_padding (self)

def	_access_token (self)

None	_send_json (self, JsonObject js)

def	_token_state (self)

def	_remove_token_state (self)

Private Attributes
	_alt_issuer

	_parameterized

	_response_code

	_params

	_test_params

Detailed Description

Core implementation of the authorization server. The API is
inheritance-based, with entry points at do_GET() and do_POST(). See the
documentation for BaseHTTPRequestHandler.

Definition at line 19 of file oauth_server.py.

Member Function Documentation

◆ _access_token()

def oauth_server.OAuthHandler._access_token ( self )

private

The actual Bearer token sent back to the client on success. Tests may
override this with the "token" test parameter.

Definition at line 221 of file oauth_server.py.

    def _access_token(self):
        """
        The actual Bearer token sent back to the client on success. Tests may
        override this with the "token" test parameter.
        """
        token = self._get_param("token", None)
        if token is not None:
            return token
 
        token = "9243959234"
        if self._alt_issuer:
            token += "-alt"
 
        return token
 

References oauth_server.OAuthHandler._alt_issuer, and oauth_server.OAuthHandler._get_param().

Referenced by oauth_server.OAuthHandler.token().

◆ _check_authn()

def oauth_server.OAuthHandler._check_authn ( self )

private

Checks the expected value of the Authorization header, if any.

Definition at line 47 of file oauth_server.py.

    def _check_authn(self):
        """
        Checks the expected value of the Authorization header, if any.
        """
        secret = self._get_param("expected_secret", None)
        if secret is None:
            return
 
        assert "Authorization" in self.headers
        method, creds = self.headers["Authorization"].split()
 
        if method != "Basic":
            raise RuntimeError(f"client used {method} auth; expected Basic")
 
        username = urllib.parse.quote_plus(self.client_id)
        password = urllib.parse.quote_plus(secret)
        expected_creds = f"{username}:{password}"
 
        if creds.encode() != base64.b64encode(expected_creds.encode()):
            raise RuntimeError(
                f"client sent '{creds}'; expected b64encode('{expected_creds}')"
            )
 

References oauth_server.OAuthHandler._get_param(), oauth_server.OAuthHandler.client_id(), printTableContent.headers, and async_ctx.headers.

◆ _check_issuer()

def oauth_server.OAuthHandler._check_issuer ( self )

private

Switches the behavior of the provider depending on the issuer URI.

Definition at line 28 of file oauth_server.py.

    def _check_issuer(self):
        """
        Switches the behavior of the provider depending on the issuer URI.
        """
        self._alt_issuer = (
            self.path.startswith("/alternate/")
            or self.path == "/.well-known/oauth-authorization-server/alternate"
        )
        self._parameterized = self.path.startswith("/param/")
 
        if self._alt_issuer:
            # The /alternate issuer uses IETF-style .well-known URIs.
            if self.path.startswith("/.well-known/"):
                self.path = self.path.removesuffix("/alternate")
            else:
                self.path = self.path.removeprefix("/alternate")
        elif self._parameterized:
            self.path = self.path.removeprefix("/param")
 

Referenced by oauth_server.OAuthHandler.do_POST().

◆ _content_type()

str oauth_server.OAuthHandler._content_type ( self )

private

Returns "application/json" unless the test has requested something
different.

Definition at line 178 of file oauth_server.py.

    def _content_type(self) -> str:
        """
        Returns "application/json" unless the test has requested something
        different.
        """
        return self._get_param("content_type", "application/json")
 

References oauth_server.OAuthHandler._get_param().

Referenced by oauth_server.OAuthHandler._send_json().

◆ _get_param()

def oauth_server.OAuthHandler._get_param	(	self,
		name,
		default
	)

private

If the client has requested a modification to this stage (see
_should_modify()), this method searches the provided test parameters for
a key of the given name, and returns it if found. Otherwise the provided
default is returned.

Definition at line 165 of file oauth_server.py.

    def _get_param(self, name, default):
        """
        If the client has requested a modification to this stage (see
        _should_modify()), this method searches the provided test parameters for
        a key of the given name, and returns it if found. Otherwise the provided
        default is returned.
        """
        if self._should_modify() and name in self._test_params:
            return self._test_params[name]
 
        return default
 

References oauth_server.OAuthHandler._should_modify(), and oauth_server.OAuthHandler._test_params.

Referenced by oauth_server.OAuthHandler._access_token(), oauth_server.OAuthHandler._check_authn(), oauth_server.OAuthHandler._content_type(), oauth_server.OAuthHandler._interval(), oauth_server.OAuthHandler._response_padding(), oauth_server.OAuthHandler._retry_code(), oauth_server.OAuthHandler._uri_spelling(), and oauth_server.OAuthHandler.token().

◆ _interval()

int oauth_server.OAuthHandler._interval ( self )

private

Returns 0 unless the test has requested something different.

Definition at line 186 of file oauth_server.py.

    def _interval(self) -> int:
        """
        Returns 0 unless the test has requested something different.
        """
        return self._get_param("interval", 0)
 

References oauth_server.OAuthHandler._get_param().

Referenced by oauth_server.OAuthHandler.authorization().

◆ _parse_params()

dict[str, str] oauth_server.OAuthHandler._parse_params ( self )

private

Parses apart the form-urlencoded request body and returns the resulting
dict. For use by do_POST().

Definition at line 86 of file oauth_server.py.

    def _parse_params(self) -> dict[str, str]:
        """
        Parses apart the form-urlencoded request body and returns the resulting
        dict. For use by do_POST().
        """
        size = int(self.headers["Content-Length"])
        form = self.rfile.read(size)
 
        assert self.headers["Content-Type"] == "application/x-www-form-urlencoded"
        return urllib.parse.parse_qs(
            form.decode("utf-8"),
            strict_parsing=True,
            keep_blank_values=True,
            encoding="utf-8",
            errors="strict",
        )
 

References oauth_server.OAuthHandler.do_POST(), printTableContent.headers, async_ctx.headers, and read.

Referenced by oauth_server.OAuthHandler.client_id().

◆ _remove_token_state()

def oauth_server.OAuthHandler._remove_token_state ( self )

private

Removes any cached _TokenState for the current client_id. Call this
after the token exchange ends to get rid of unnecessary state.

Definition at line 284 of file oauth_server.py.

    def _remove_token_state(self):
        """
        Removes any cached _TokenState for the current client_id. Call this
        after the token exchange ends to get rid of unnecessary state.
        """
        if self.client_id in self.server.token_state:
            del self.server.token_state[self.client_id]
 

References oauth_server.OAuthHandler.client_id(), and PgFdwRelationInfo.server.

Referenced by oauth_server.OAuthHandler.token().

◆ _response_padding()

def oauth_server.OAuthHandler._response_padding ( self )

private

If the huge_response test parameter is set to True, returns a dict
containing a gigantic string value, which can then be folded into a JSON
response.

Definition at line 209 of file oauth_server.py.

    def _response_padding(self):
        """
        If the huge_response test parameter is set to True, returns a dict
        containing a gigantic string value, which can then be folded into a JSON
        response.
        """
        if not self._get_param("huge_response", False):
            return dict()
 
        return {"_pad_": "x" * 1024 * 1024}
 

References oauth_server.OAuthHandler._get_param().

Referenced by oauth_server.OAuthHandler.authorization(), and oauth_server.OAuthHandler.token().

◆ _retry_code()

str oauth_server.OAuthHandler._retry_code ( self )

private

Returns "authorization_pending" unless the test has requested something
different.

Definition at line 193 of file oauth_server.py.

    def _retry_code(self) -> str:
        """
        Returns "authorization_pending" unless the test has requested something
        different.
        """
        return self._get_param("retry_code", "authorization_pending")
 

References oauth_server.OAuthHandler._get_param().

Referenced by oauth_server.OAuthHandler.token().

◆ _send_json()

None oauth_server.OAuthHandler._send_json	(		self,
		JsonObject	js
	)

private

Sends the provided JSON dict as an application/json response.
self._response_code can be modified to send JSON error responses.

Definition at line 236 of file oauth_server.py.

    def _send_json(self, js: JsonObject) -> None:
        """
        Sends the provided JSON dict as an application/json response.
        self._response_code can be modified to send JSON error responses.
        """
        resp = json.dumps(js).encode("ascii")
        self.log_message("sending JSON response: %s", resp)
 
        self.send_response(self._response_code)
        self.send_header("Content-Type", self._content_type)
        self.send_header("Content-Length", str(len(resp)))
        self.end_headers()
 
        self.wfile.write(resp)
 

References oauth_server.OAuthHandler._content_type(), oauth_server.OAuthHandler._response_code, len, str, and write.

◆ _should_modify()

bool oauth_server.OAuthHandler._should_modify ( self )

private

Returns True if the client has requested a modification to this stage of
the exchange.

Definition at line 145 of file oauth_server.py.

    def _should_modify(self) -> bool:
        """
        Returns True if the client has requested a modification to this stage of
        the exchange.
        """
        if not hasattr(self, "_test_params"):
            return False
 
        stage = self._test_params.get("stage")
 
        return (
            stage == "all"
            or (
                stage == "discovery"
                and self.path == "/.well-known/openid-configuration"
            )
            or (stage == "device" and self.path == "/authorize")
            or (stage == "token" and self.path == "/token")
        )
 

References oauth_server.OAuthHandler._test_params, RewriteMappingFile.path, backup_file_entry.path, PathClauseUsage.path, JsonTablePlanState.path, keepwal_entry.path, file_entry_t.path, fetch_range_request.path, UpgradeTaskReport.path, tablespaceinfo.path, IndexPath.path, BitmapHeapPath.path, BitmapAndPath.path, BitmapOrPath.path, TidPath.path, TidRangePath.path, SubqueryScanPath.path, ForeignPath.path, CustomPath.path, AppendPath.path, MergeAppendPath.path, GroupResultPath.path, MaterialPath.path, MemoizePath.path, UniquePath.path, GatherPath.path, GatherMergePath.path, ProjectionPath.path, ProjectSetPath.path, SortPath.path, GroupPath.path, UpperUniquePath.path, AggPath.path, GroupingSetsPath.path, MinMaxAggPath.path, WindowAggPath.path, SetOpPath.path, RecursiveUnionPath.path, LockRowsPath.path, ModifyTablePath.path, LimitPath.path, MinMaxAggInfo.path, JsonTablePathScan.path, _include_path.path, and oauth_server.OAuthHandler.path.

Referenced by oauth_server.OAuthHandler._get_param(), and oauth_server.OAuthHandler.token().

◆ _token_state()

def oauth_server.OAuthHandler._token_state ( self )

private

A cached _TokenState object for the connected client (as determined by
the request's client_id), or a new one if it doesn't already exist.

This relies on the existence of a defaultdict attached to the server;
see main() below.

Definition at line 274 of file oauth_server.py.

    def _token_state(self):
        """
        A cached _TokenState object for the connected client (as determined by
        the request's client_id), or a new one if it doesn't already exist.
 
        This relies on the existence of a defaultdict attached to the server;
        see main() below.
        """
        return self.server.token_state[self.client_id]
 

References oauth_server.OAuthHandler.client_id(), oauth_server.main(), and PgFdwRelationInfo.server.

Referenced by oauth_server.OAuthHandler.authorization(), and oauth_server.OAuthHandler.token().

◆ _uri_spelling()

str oauth_server.OAuthHandler._uri_spelling ( self )

private

Returns "verification_uri" unless the test has requested something
different.

Definition at line 201 of file oauth_server.py.

    def _uri_spelling(self) -> str:
        """
        Returns "verification_uri" unless the test has requested something
        different.
        """
        return self._get_param("uri_spelling", "verification_uri")
 

References oauth_server.OAuthHandler._get_param().

Referenced by oauth_server.OAuthHandler.authorization().

◆ authorization()

JsonObject oauth_server.OAuthHandler.authorization ( self )

Definition at line 292 of file oauth_server.py.

    def authorization(self) -> JsonObject:
        uri = "https://example.com/"
        if self._alt_issuer:
            uri = "https://example.org/"
 
        resp = {
            "device_code": "postgres",
            "user_code": "postgresuser",
            self._uri_spelling: uri,
            "expires_in": 5,
            **self._response_padding,
        }
 
        interval = self._interval
        if interval is not None:
            resp["interval"] = interval
            self._token_state.min_delay = interval
        else:
            self._token_state.min_delay = 5  # default
 
        # Check the scope.
        if "scope" in self._params:
            assert self._params["scope"][0], "empty scopes should be omitted"
 
        return resp
 

References oauth_server.OAuthHandler._alt_issuer, oauth_server.OAuthHandler._interval(), oauth_server.OAuthHandler._params, oauth_server.OAuthHandler._response_padding(), oauth_server.OAuthHandler._token_state(), and oauth_server.OAuthHandler._uri_spelling().

◆ client_id()

str oauth_server.OAuthHandler.client_id ( self )

Returns the client_id sent in the POST body or the Authorization header.
self._parse_params() must have been called first.

Definition at line 104 of file oauth_server.py.

    def client_id(self) -> str:
        """
        Returns the client_id sent in the POST body or the Authorization header.
        self._parse_params() must have been called first.
        """
        if "client_id" in self._params:
            return self._params["client_id"][0]
 
        if "Authorization" not in self.headers:
            raise RuntimeError("client did not send any client_id")
 
        _, creds = self.headers["Authorization"].split()
 
        decoded = base64.b64decode(creds).decode("utf-8")
        username, _ = decoded.split(":", 1)
 
        return urllib.parse.unquote_plus(username)
 

References oauth_server.OAuthHandler._params, oauth_server.OAuthHandler._parse_params(), printTableContent.headers, and async_ctx.headers.

Referenced by oauth_server.OAuthHandler._check_authn(), oauth_server.OAuthHandler._remove_token_state(), and oauth_server.OAuthHandler._token_state().

◆ config()

JsonObject oauth_server.OAuthHandler.config ( self )

Definition at line 251 of file oauth_server.py.

    def config(self) -> JsonObject:
        port = self.server.socket.getsockname()[1]
 
        issuer = f"http://localhost:{port}"
        if self._alt_issuer:
            issuer += "/alternate"
        elif self._parameterized:
            issuer += "/param"
 
        return {
            "issuer": issuer,
            "token_endpoint": issuer + "/token",
            "device_authorization_endpoint": issuer + "/authorize",
            "response_types_supported": ["token"],
            "subject_types_supported": ["public"],
            "id_token_signing_alg_values_supported": ["RS256"],
            "grant_types_supported": [
                "authorization_code",
                "urn:ietf:params:oauth:grant-type:device_code",
            ],
        }
 

References oauth_server.OAuthHandler._alt_issuer, oauth_server.OAuthHandler._parameterized, and PgFdwRelationInfo.server.

◆ do_GET()

def oauth_server.OAuthHandler.do_GET ( self )

Definition at line 70 of file oauth_server.py.

    def do_GET(self):
        self._response_code = 200
        self._check_issuer()
 
        config_path = "/.well-known/openid-configuration"
        if self._alt_issuer:
            config_path = "/.well-known/oauth-authorization-server"
 
        if self.path == config_path:
            resp = self.config()
        else:
            self.send_error(404, "Not Found")
            return
 
        self._send_json(resp)
 

◆ do_POST()

def oauth_server.OAuthHandler.do_POST ( self )

Definition at line 122 of file oauth_server.py.

    def do_POST(self):
        self._response_code = 200
        self._check_issuer()
 
        self._params = self._parse_params()
        if self._parameterized:
            # Pull encoded test parameters out of the peer's client_id field.
            # This is expected to be Base64-encoded JSON.
            js = base64.b64decode(self.client_id)
            self._test_params = json.loads(js)
 
        self._check_authn()
 
        if self.path == "/authorize":
            resp = self.authorization()
        elif self.path == "/token":
            resp = self.token()
        else:
            self.send_error(404)
            return
 
        self._send_json(resp)
 

References oauth_server.OAuthHandler._check_issuer(), and oauth_server.OAuthHandler._response_code.

Referenced by oauth_server.OAuthHandler._parse_params().

◆ token()

JsonObject oauth_server.OAuthHandler.token ( self )

Definition at line 318 of file oauth_server.py.

    def token(self) -> JsonObject:
        if err := self._get_param("error_code", None):
            self._response_code = self._get_param("error_status", 400)
 
            resp = {"error": err}
            if desc := self._get_param("error_desc", ""):
                resp["error_description"] = desc
 
            return resp
 
        if self._should_modify() and "retries" in self._test_params:
            retries = self._test_params["retries"]
 
            # Check to make sure the token interval is being respected.
            now = time.monotonic()
            if self._token_state.last_try is not None:
                delay = now - self._token_state.last_try
                assert (
                    delay > self._token_state.min_delay
                ), f"client waited only {delay} seconds between token requests (expected {self._token_state.min_delay})"
 
            self._token_state.last_try = now
 
            # If we haven't reached the required number of retries yet, return a
            # "pending" response.
            if self._token_state.retries < retries:
                self._token_state.retries += 1
 
                self._response_code = 400
                return {"error": self._retry_code}
 
        # Clean up any retry tracking state now that the exchange is ending.
        self._remove_token_state()
 
        return {
            "access_token": self._access_token,
            "token_type": "bearer",
            **self._response_padding,
        }
 
 

References oauth_server.OAuthHandler._access_token(), oauth_server.OAuthHandler._get_param(), oauth_server.OAuthHandler._remove_token_state(), oauth_server.OAuthHandler._response_code, oauth_server.OAuthHandler._response_padding(), oauth_server.OAuthHandler._retry_code(), oauth_server.OAuthHandler._should_modify(), oauth_server.OAuthHandler._test_params, and oauth_server.OAuthHandler._token_state().

Field Documentation

◆ _alt_issuer

oauth_server.OAuthHandler._alt_issuer

private

Definition at line 32 of file oauth_server.py.

Referenced by oauth_server.OAuthHandler._access_token(), oauth_server.OAuthHandler.authorization(), and oauth_server.OAuthHandler.config().

◆ _parameterized

oauth_server.OAuthHandler._parameterized

private

Definition at line 36 of file oauth_server.py.

Referenced by oauth_server.OAuthHandler.config().

◆ _params

oauth_server.OAuthHandler._params

private

Definition at line 126 of file oauth_server.py.

Referenced by oauth_server.OAuthHandler.authorization(), and oauth_server.OAuthHandler.client_id().

◆ _response_code

oauth_server.OAuthHandler._response_code

private

Definition at line 71 of file oauth_server.py.

Referenced by oauth_server.OAuthHandler._send_json(), oauth_server.OAuthHandler.do_POST(), and oauth_server.OAuthHandler.token().

◆ _test_params

oauth_server.OAuthHandler._test_params

private

Definition at line 131 of file oauth_server.py.

Referenced by oauth_server.OAuthHandler._get_param(), oauth_server.OAuthHandler._should_modify(), and oauth_server.OAuthHandler.token().

◆ JsonObject

oauth_server.OAuthHandler.JsonObject = dict[str, object]

static

Definition at line 26 of file oauth_server.py.

◆ path

oauth_server.OAuthHandler.path

Definition at line 34 of file oauth_server.py.

Referenced by oauth_server.OAuthHandler._should_modify().

The documentation for this class was generated from the following file:

src/test/modules/oauth_validator/t/oauth_server.py

Public Member Functions

Data Fields

Static Public Attributes

Private Member Functions

Private Attributes

Detailed Description

Member Function Documentation

◆ _access_token()

◆ _check_authn()

◆ _check_issuer()

◆ _content_type()

◆ _get_param()

◆ _interval()

◆ _parse_params()

◆ _remove_token_state()

◆ _response_padding()

◆ _retry_code()

◆ _send_json()

◆ _should_modify()

◆ _token_state()

◆ _uri_spelling()

◆ authorization()

◆ client_id()

◆ config()

◆ do_GET()

◆ do_POST()

◆ token()

Field Documentation

◆ _alt_issuer

◆ _parameterized

◆ _params

◆ _response_code

◆ _test_params

◆ JsonObject

◆ path