-
Notifications
You must be signed in to change notification settings - Fork 3
Expand file tree
/
Copy pathgenerate_ci.py
More file actions
63 lines (51 loc) · 1.78 KB
/
Copy pathgenerate_ci.py
File metadata and controls
63 lines (51 loc) · 1.78 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
import json
import re
import time
import jwt
from utils import get_access_token_jwt_secret, bad_request, is_allowed_domain
from aws_lambda_powertools import Logger
from aws_lambda_powertools.utilities.typing import LambdaContext
logger = Logger()
@logger.inject_lambda_context
def handler(event, context: LambdaContext) -> dict:
request_ip = event['requestContext']['identity']['sourceIp']
logger.append_keys(request_id=context.aws_request_id, request_ip=request_ip)
assert event['httpMethod'] == 'POST'
logger.info({"message": "Processing POST request", "body": event})
caller = event['requestContext']['identity']['caller']
values = json.loads(event['body'])
try:
exp_in = int(values['exp_in'])
subject = values['subject']
assert len(subject) > 0
except (KeyError, AssertionError):
return bad_request('mandatory fields not present')
if exp_in > (60 * 60 * 6):
return bad_request('expiration too long (max 6 hours)')
domains = set(values['domains'])
for domain in domains:
if not re.match(r'^[a-zA-Z0-9.-]+$', domain):
return bad_request('', f"`{domain}` does not look like a domain name")
if not is_allowed_domain(domain):
return bad_request('', 'Unknown domain in request')
now = int(time.time())
ci_token = {
'iat': now,
'exp': now + exp_in,
'azp': caller,
'domains': list(domains),
'sub': subject,
}
logger.info({"message": "Issuing JWT", "jwt": ci_token})
raw_ci_token = jwt.encode(
ci_token,
get_access_token_jwt_secret(),
algorithm='HS256',
)
return {
'statusCode': 200,
'headers': {
'Content-Type': 'text/plain',
},
'body': raw_ci_token,
}