app/doorboy-proxy.py

@@ -5,6 +5,7 @@ from sanic.response import text, json
 from sanic_prometheus import monitor
 from dateutil.parser import parse
 import httpx
+from functools import wraps
 from motor.motor_asyncio import AsyncIOMotorClient
 from pymongo.errors import PyMongoError
 import os
@@ -32,12 +33,23 @@ async def setup_db(app, loop):
     # https://github.com/sanic-org/sanic/issues/919
     app.ctx.db = AsyncIOMotorClient(MONGO_URI).get_default_database()
 
-@app.route("/allowed")
-async def view_doorboy_uids(request):
-    key = request.headers.get("KEY")
-    if not key or key not in [DOORBOY_SECRET_FLOOR, DOORBOY_SECRET_WORKSHOP]:
-        return text("how about no")
+def authenticate_door(wrapped):
+    def decorator(f):
+        @wraps(f)
+        async def decorated_function(request, *args, **kwargs):
+            doorboy_secret = request.headers.get("KEY")
+            if doorboy_secret not in [DOORBOY_SECRET_FLOOR, DOORBOY_SECRET_WORKSHOP]:
+                return text("Invalid doorboy secret token", status=401)
+            
+            return await f(request, *args, **kwargs)
+        return decorated_function
+    return decorator(wrapped)
 
+@app.route("/allowed")
+@authenticate_door
+async def view_doorboy_uids(request):
+    # authorize
+    key = request.headers.get("KEY")
     groups = []
     if key == DOORBOY_SECRET_FLOOR:
         groups.append(FLOOR_ACCESS_GROUP)
@@ -101,12 +113,8 @@ async def view_open_door_events(request):
     return json(transformed, default=datetime_to_json_formatting)
 
 @app.route("/longpoll", stream=True)
+@authenticate_door
 async def view_longpoll(request):
-    key = request.headers.get("KEY")
-    if not key or key not in [DOORBOY_SECRET_FLOOR, DOORBOY_SECRET_WORKSHOP]:
-        return text("Invalid token")
-    
-    # authenticate
     response = await request.respond(content_type="text/event-stream")
     await response.send("data: response-generator-started\n\n")
     pipeline = [
@@ -134,13 +142,10 @@ async def view_longpoll(request):
 
 # Called by the door to log a card swipe. Does not decide whether the door should be opened.
 @app.post("/swipe")
+@authenticate_door
 async def swipe(request):
-    # authenticate
-    key = request.headers.get("KEY")
-    if not key or key not in [DOORBOY_SECRET_FLOOR, DOORBOY_SECRET_WORKSHOP]:
-        return text("Invalid token", status=401)
-    
     # authorize
+    key = request.headers.get("KEY")
     data = request.json
     doors = set()
     if key == DOORBOY_SECRET_FLOOR: