Initial commit: boarding pass to AirTrail pipeline

Photograph/screenshot a boarding pass, decode its BCBP barcode (PDF417/
Aztec/QR/DataMatrix), fetch the historical flight track from FlightAware
the day after the flight, stage it for review, and write it into AirTrail
via its REST API on approval. ntfy notifications carry signed
Approve/Deny/Investigate actions.
This commit is contained in:
2026-07-04 22:21:02 +10:00
commit 408fde440d
34 changed files with 11657 additions and 0 deletions
+43
View File
@@ -0,0 +1,43 @@
"""Signed tokens authorizing the ntfy Approve/Deny callback URLs.
ntfy action buttons fire a plain, unauthenticated HTTP request from wherever
the notification is received - they carry no Authelia session. The reverse
proxy bypass needed for those routes (see README) applies to the whole path,
so this token is the ONLY auth on /flights/<id>/approve and /reject,
including when the request comes from the review page's own buttons.
"""
import hashlib
import hmac
import time
import models
def _secret() -> bytes:
return models.get_or_create_secret_key().encode("utf-8")
def _signature(flight_id: int, action: str, expiry: int) -> str:
message = f"{flight_id}:{action}:{expiry}".encode("utf-8")
return hmac.new(_secret(), message, hashlib.sha256).hexdigest()
def sign_token(flight_id: int, action: str, ttl_seconds: int = 30 * 86400) -> str:
expiry = int(time.time()) + ttl_seconds
sig = _signature(flight_id, action, expiry)
return f"{expiry}.{sig}"
def verify_token(flight_id: int, action: str, token: str) -> bool:
if not token or "." not in token:
return False
expiry_str, _, sig = token.partition(".")
try:
expiry = int(expiry_str)
except ValueError:
return False
if expiry < time.time():
return False
expected = _signature(flight_id, action, expiry)
return hmac.compare_digest(expected, sig)