birminghack-checkin/main.py at main · CSSUoB/birminghack-checkin · GitHub

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
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
from fastapi import Request, Response, FastAPI
from strictyaml import load
from proxy import start_reverse_tunnel
import threading
import logging
import hmac
import hashlib
import printer
import imggen
import base64
import json
import uvicorn

with open("config.yaml", "r") as f:
    config = load(f.read()).data

logger = logging.getLogger('uvicorn.error')
lock = threading.Lock()

printer.init_printer(
    int(config['printer']['maj'], 16),
    int(config['printer']['min'], 16)
)

app = FastAPI()

group = 1
serial = 1

def verify_message(message: bytes, signature: bytes):
    h = hmac.new(config['tito']['mac_token'].encode('utf-8'), message, hashlib.sha256).digest()
    return hmac.compare_digest(h, signature)

@app.post("/")
async def create_checkin(request: Request):
    body = await request.body()
    logger.debug(f"Received message: {body.decode('utf-8')}")
    signature = request.headers.get("Tito-Signature")
    if signature is None:
        logger.warning("Rejecting message with missing signature")
        return Response(status_code=403)

    if not verify_message(body, base64.b64decode(signature)):
        logger.warning("Rejecting unauthenticated message")
        return Response(status_code=403)

    lock.acquire(True)

    try:
        global group
        global serial

        content = json.loads(body.decode('utf-8'))
        attendee_name = content['name']
        ticket_type = content['release_title']
        ticket_ref = content['reference']
        slug = content['slug']
        pronouns = list(filter(lambda a: a['question'] == 'What are your preferred pronouns?', content['answers']))[0]['response']
        logger.info(f"Printing attendee pass for {attendee_name}...")
        printer.print_pass(imggen.name(attendee_name), imggen.pronouns(pronouns), ticket_ref, ticket_type, slug)

        pizza_pref = list(filter(lambda a: a['question'] == 'What is your pizza preference?', content['answers']))
        if len(pizza_pref) == 0:
            return Response(status_code=204)

        d_reqs = None
        q_d_reqs = list(filter(lambda a: a['question'] == 'Do you have any dietary restrictions?', content['answers']))
        if len(q_d_reqs) > 0:
            d_reqs = q_d_reqs[0]['response']

        logger.info(f"Printing food token for {attendee_name}...")
        printer.print_food(attendee_name, pizza_pref[0]['response'], str(group), d_reqs)

        serial = serial + 1
        if serial % 10 == 0:
            group = group + 1
            logger.info(f"Incrementing group counter to {group}")

        return Response(status_code=204)
    finally:
        lock.release()

if __name__ == '__main__':
    if 'ssh' in config:
        threading.Thread(target=start_reverse_tunnel, args=(config,), daemon=True).start()

    uvicorn.run(app, log_level="info", host="0.0.0.0", port=3000)