database/postgres
1
0
Fork 0
mirror of https://github.com/postgres/postgres.git synced 2025-06-14 18:42:34 +03:00
Code Activity

oauth: Support Python 3.6 in tests

Browse Source 
RHEL8 ships a patched 3.6.8 as its base Python version, and I
accidentally let some newer Python-isms creep into oauth_server.py
during development.

Reported-by: Tom Lane <tgl@sss.pgh.pa.us>
Reviewed-by: Jelte Fennema-Nio <postgres@jeltef.nl>
Tested-by: Renan Alves Fonseca <renanfonseca@gmail.com>
Tested-by: Tom Lane <tgl@sss.pgh.pa.us>
Discussion: https://postgr.es/m/16098.1745079444%40sss.pgh.pa.us
This commit is contained in:
Jacob Champion
2025-04-23 11:16:45 -07:00
parent bb78e42678
commit 005ccae0f2
1 changed files with 17 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
src/test/modules/oauth_validator/t/oauth_server.py
View File

@ -14,6 +14,7 @@ import sys
import time import time
import urllib.parse import urllib.parse
from collections import defaultdict from collections import defaultdict
from typing import Dict
class OAuthHandler(http.server.BaseHTTPRequestHandler): class OAuthHandler(http.server.BaseHTTPRequestHandler):
@ -23,7 +24,7 @@ class OAuthHandler(http.server.BaseHTTPRequestHandler):
documentation for BaseHTTPRequestHandler. documentation for BaseHTTPRequestHandler.
""" """
JsonObject = dict[str, object] # TypeAlias is not available until 3.10 JsonObject = Dict[str, object] # TypeAlias is not available until 3.10
def _check_issuer(self): def _check_issuer(self):
""" """
@ -35,14 +36,16 @@ class OAuthHandler(http.server.BaseHTTPRequestHandler):
) )
self._parameterized = self.path.startswith("/param/") self._parameterized = self.path.startswith("/param/")
# Strip off the magic path segment. (The more readable
# str.removeprefix()/removesuffix() aren't available until Py3.9.)
if self._alt_issuer: if self._alt_issuer:
# The /alternate issuer uses IETF-style .well-known URIs. # The /alternate issuer uses IETF-style .well-known URIs.
if self.path.startswith("/.well-known/"): if self.path.startswith("/.well-known/"):
self.path = self.path.removesuffix("/alternate") self.path = self.path[: -len("/alternate")]
else: else:
self.path = self.path.removeprefix("/alternate") self.path = self.path[len("/alternate") :]
elif self._parameterized: elif self._parameterized:
self.path = self.path.removeprefix("/param") self.path = self.path[len("/param") :]
def _check_authn(self): def _check_authn(self):
""" """
@ -58,8 +61,10 @@ class OAuthHandler(http.server.BaseHTTPRequestHandler):
if method != "Basic": if method != "Basic":
raise RuntimeError(f"client used {method} auth; expected Basic") raise RuntimeError(f"client used {method} auth; expected Basic")
username = urllib.parse.quote_plus(self.client_id) # TODO: Remove "~" from the safe list after Py3.6 support is removed.
password = urllib.parse.quote_plus(secret) # 3.7 does this by default.
username = urllib.parse.quote_plus(self.client_id, safe="~")
password = urllib.parse.quote_plus(secret, safe="~")
expected_creds = f"{username}:{password}" expected_creds = f"{username}:{password}"
if creds.encode() != base64.b64encode(expected_creds.encode()): if creds.encode() != base64.b64encode(expected_creds.encode()):
@ -83,7 +88,7 @@ class OAuthHandler(http.server.BaseHTTPRequestHandler):
self._send_json(resp) self._send_json(resp)
def _parse_params(self) -> dict[str, str]: def _parse_params(self) -> Dict[str, str]:
""" """
Parses apart the form-urlencoded request body and returns the resulting Parses apart the form-urlencoded request body and returns the resulting
dict. For use by do_POST(). dict. For use by do_POST().
@ -316,11 +321,14 @@ class OAuthHandler(http.server.BaseHTTPRequestHandler):
return resp return resp
def token(self) -> JsonObject: def token(self) -> JsonObject:
if err := self._get_param("error_code", None): err = self._get_param("error_code", None)
if err:
self._response_code = self._get_param("error_status", 400) self._response_code = self._get_param("error_status", 400)
resp = {"error": err} resp = {"error": err}
if desc := self._get_param("error_desc", ""):
desc = self._get_param("error_desc", "")
if desc:
resp["error_description"] = desc resp["error_description"] = desc
return resp return resp