|
| 1 | +"""Provides helper functions for interacting with the Passage Identity API.""" |
| 2 | + |
1 | 3 | import re |
| 4 | +from http import HTTPStatus |
| 5 | + |
| 6 | +from requests.sessions import Request |
2 | 7 |
|
3 | 8 | from passageidentity import requests |
4 | 9 | from passageidentity.errors import PassageError |
5 | 10 |
|
6 | | -TOKEN_TYPE = "Bearer" |
| 11 | +BEARER_PATTERN = r"Bearer ([^\s,]+)" |
7 | 12 | BASE_URL = "https://api.passage.id/v1/apps/" |
8 | 13 |
|
9 | | -""" |
10 | | -Helper function to extract the JWT from an Authorization header. |
11 | | -""" |
12 | 14 |
|
| 15 | +def extract_token(auth_header: str) -> str: |
| 16 | + """Extract the JWT from an Authorization header.""" |
| 17 | + expression = re.escape(BEARER_PATTERN) |
| 18 | + match = re.search(expression, auth_header) |
13 | 19 |
|
14 | | -def extractToken(authHeader): |
15 | | - expression = re.escape(TOKEN_TYPE) + r" ([^\s,]+)" |
16 | | - match = re.search(expression, authHeader) |
17 | | - try: |
| 20 | + if match: |
18 | 21 | return match.group(1) |
19 | | - except (AttributeError, IndexError): |
20 | | - raise PassageError("No Passage authorization header.") |
21 | 22 |
|
| 23 | + msg = "No Passage authorization header." |
| 24 | + raise PassageError(msg) |
22 | 25 |
|
23 | | -""" |
24 | | -Helper funtion to get the auth token from a request. |
25 | | -Checks the Authorization header first, then the psg_auth_token cookie |
26 | | -""" |
27 | 26 |
|
| 27 | +def get_auth_token_from_request(request: Request, auth_strategy: int) -> str: |
| 28 | + """Get the auth token from a request. |
28 | 29 |
|
29 | | -def getAuthTokenFromRequest(request, auth_strategy): |
30 | | - if auth_strategy == 2: |
31 | | - authHeader = request.headers["Authorization"] |
32 | | - expression = re.escape(TOKEN_TYPE) + r" ([^\s,]+)" |
33 | | - match = re.search(expression, authHeader) |
34 | | - try: |
| 30 | + Checks the Authorization header first, then the psg_auth_token cookie. |
| 31 | + """ |
| 32 | + if auth_strategy == 2: # noqa: PLR2004 |
| 33 | + auth_header = request.headers["Authorization"] |
| 34 | + expression = re.escape(BEARER_PATTERN) |
| 35 | + match = re.search(expression, auth_header) |
| 36 | + |
| 37 | + if match: |
35 | 38 | return match.group(1) |
36 | | - except (AttributeError, IndexError): |
37 | | - raise PassageError("No Passage authorization header.") |
38 | | - else: |
39 | | - try: |
40 | | - cookies = request.COOKIES |
41 | | - if "psg_auth_token" not in cookies.keys(): |
42 | | - raise PassageError("No Passage authentication token.") |
43 | | - return cookies["psg_auth_token"] |
44 | | - except Exception: |
45 | | - try: |
46 | | - cookies = request.cookies |
47 | | - if "psg_auth_token" not in cookies.keys(): |
48 | | - raise PassageError("No Passage authentication token.") |
49 | | - return cookies["psg_auth_token"] |
50 | | - except Exception: |
51 | | - raise PassageError("No passage authentication token") |
52 | | - |
53 | | - |
54 | | -""" |
55 | | -Helper function to fetch the public key for the given app id from Passage |
56 | | -""" |
57 | | - |
58 | | - |
59 | | -def fetchApp(app_id): |
| 39 | + |
| 40 | + msg = "No Passage authorization header." |
| 41 | + raise PassageError(msg) |
| 42 | + |
| 43 | + if "psg_auth_token" not in request.cookies: |
| 44 | + msg = "No Passage authentication token." |
| 45 | + raise PassageError(msg) |
| 46 | + |
| 47 | + return request.cookies["psg_auth_token"] |
| 48 | + |
| 49 | + |
| 50 | +def fetch_app(app_id: str) -> dict: |
| 51 | + """Fetch the public key for the given app id from Passage.""" |
60 | 52 | # unauthenticated request to get the public key |
61 | 53 | r = requests.get(BASE_URL + app_id) |
62 | 54 |
|
63 | | - # check response code |
64 | | - if r.status_code != 200: |
| 55 | + if r.status_code != HTTPStatus.OK: |
65 | 56 | raise PassageError("Could not fetch app information for app id " + app_id) |
66 | 57 |
|
67 | 58 | return r.json()["app"] |
0 commit comments