Skip to content

lablink_allocator_service.signed_cookie

HMAC-SHA256 signed cookie value helpers.

Cookie format: .

Bases: Exception

Read cookie_signing_secret from settings; create + persist on first use.

conn is a psycopg2 connection. Caller owns the connection lifecycle.

Source code in packages/allocator/src/lablink_allocator_service/signed_cookie.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
def get_or_create_cookie_secret(conn) -> str:
    """Read cookie_signing_secret from settings; create + persist on first use.

    `conn` is a psycopg2 connection. Caller owns the connection lifecycle.
    """
    with conn.cursor() as cur:
        cur.execute(
            "SELECT value FROM settings WHERE key = 'cookie_signing_secret'"
        )
        row = cur.fetchone()
        if row is not None:
            return row[0]
        new_secret = _secrets.token_urlsafe(32)
        cur.execute(
            "INSERT INTO settings (key, value) "
            "VALUES ('cookie_signing_secret', %s) "
            "ON CONFLICT (key) DO NOTHING",
            (new_secret,),
        )
        conn.commit()
        cur.execute(
            "SELECT value FROM settings WHERE key = 'cookie_signing_secret'"
        )
        return cur.fetchone()[0]
Source code in packages/allocator/src/lablink_allocator_service/signed_cookie.py 
25
26
27
28
29
30
def sign(payload: str, *, secret: str) -> str:
    payload_bytes = payload.encode("utf-8")
    mac = hmac.new(
        secret.encode("utf-8"), payload_bytes, hashlib.sha256
    ).digest()
    return f"{_b64encode(payload_bytes)}.{_b64encode(mac)}"
Source code in packages/allocator/src/lablink_allocator_service/signed_cookie.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
def verify(token: str, *, secret: str) -> str:
    try:
        payload_b64, mac_b64 = token.split(".", 1)
        payload_bytes = _b64decode(payload_b64)
        provided_mac = _b64decode(mac_b64)
    except (ValueError, binascii.Error) as exc:
        raise InvalidSignature("malformed token") from exc
    expected_mac = hmac.new(
        secret.encode("utf-8"), payload_bytes, hashlib.sha256
    ).digest()
    if not hmac.compare_digest(provided_mac, expected_mac):
        raise InvalidSignature("hmac mismatch")
    return payload_bytes.decode("utf-8")