Spaces:
Sleeping
Sleeping
Update main.py
Browse files
main.py
CHANGED
|
@@ -181,70 +181,82 @@ def send_rotation_notification(job_id, shift_record):
|
|
| 181 |
except Exception as e:
|
| 182 |
logger.error(f"Error sending rotation notification for job {job_id}: {e}")
|
| 183 |
|
|
|
|
|
|
|
|
|
|
|
|
|
| 184 |
# === Auth Middleware ===
|
| 185 |
def verify_token(req):
|
| 186 |
"""
|
| 187 |
Verify Firebase ID token and check if user is admin.
|
|
|
|
| 188 |
Returns decoded token dict if valid admin, None otherwise.
|
| 189 |
"""
|
| 190 |
auth_header = req.headers.get("Authorization")
|
| 191 |
-
logger.debug(f"Verifying token for request to {req.path}. Authorization header present: {bool(auth_header)}")
|
| 192 |
-
|
| 193 |
if not auth_header:
|
| 194 |
-
logger.warning("Authorization header missing.")
|
| 195 |
return None
|
| 196 |
|
| 197 |
try:
|
| 198 |
# Extract token
|
| 199 |
if auth_header.startswith("Bearer "):
|
| 200 |
token = auth_header[7:].strip()
|
| 201 |
-
logger.debug("Bearer prefix found in Authorization header.")
|
| 202 |
else:
|
| 203 |
token = auth_header.strip()
|
| 204 |
-
logger.debug("No Bearer prefix found, using header as token.")
|
| 205 |
-
|
| 206 |
-
if not token:
|
| 207 |
-
logger.warning("Authorization header is empty or only whitespace.")
|
| 208 |
-
return None
|
| 209 |
|
| 210 |
-
logger.debug("Attempting to verify Firebase ID token...")
|
| 211 |
# Verify the token
|
| 212 |
decoded = firebase_auth.verify_id_token(token)
|
| 213 |
-
|
| 214 |
-
|
| 215 |
-
# Get user UID from decoded token
|
| 216 |
uid = decoded.get('uid')
|
|
|
|
|
|
|
| 217 |
if not uid:
|
| 218 |
-
|
| 219 |
return None
|
| 220 |
|
| 221 |
-
|
| 222 |
-
# Check if user is admin by querying Firebase Realtime Database
|
| 223 |
admins_ref = db.reference("/admins")
|
| 224 |
admin_data = admins_ref.child(uid).get()
|
| 225 |
-
logger.debug(f"Admin data retrieved for UID {uid}: {admin_data}")
|
| 226 |
|
| 227 |
if admin_data and admin_data.get("is_admin", False):
|
| 228 |
-
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 229 |
return decoded
|
| 230 |
else:
|
| 231 |
-
|
|
|
|
| 232 |
return None
|
| 233 |
-
|
| 234 |
except firebase_auth.InvalidIdTokenError as e:
|
| 235 |
-
|
| 236 |
return None
|
| 237 |
except firebase_auth.ExpiredIdTokenError as e:
|
| 238 |
-
|
| 239 |
return None
|
| 240 |
except firebase_auth.RevokedIdTokenError as e:
|
| 241 |
-
|
| 242 |
return None
|
| 243 |
except Exception as e:
|
| 244 |
-
|
|
|
|
|
|
|
|
|
|
| 245 |
return None
|
| 246 |
|
| 247 |
|
|
|
|
| 248 |
# === Admin Setup (Legacy - kept for compatibility) ===
|
| 249 |
def setup_admins():
|
| 250 |
ref = db.reference("admins")
|
|
|
|
| 181 |
except Exception as e:
|
| 182 |
logger.error(f"Error sending rotation notification for job {job_id}: {e}")
|
| 183 |
|
| 184 |
+
# === Auth Middleware ===
|
| 185 |
+
# Add this import at the top if not already there
|
| 186 |
+
# from firebase_admin import auth as firebase_auth # Already imported
|
| 187 |
+
|
| 188 |
# === Auth Middleware ===
|
| 189 |
def verify_token(req):
|
| 190 |
"""
|
| 191 |
Verify Firebase ID token and check if user is admin.
|
| 192 |
+
Automatically creates admin entry in DB if email is in ADMIN_EMAILS but UID is not found.
|
| 193 |
Returns decoded token dict if valid admin, None otherwise.
|
| 194 |
"""
|
| 195 |
auth_header = req.headers.get("Authorization")
|
|
|
|
|
|
|
| 196 |
if not auth_header:
|
|
|
|
| 197 |
return None
|
| 198 |
|
| 199 |
try:
|
| 200 |
# Extract token
|
| 201 |
if auth_header.startswith("Bearer "):
|
| 202 |
token = auth_header[7:].strip()
|
|
|
|
| 203 |
else:
|
| 204 |
token = auth_header.strip()
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 205 |
|
|
|
|
| 206 |
# Verify the token
|
| 207 |
decoded = firebase_auth.verify_id_token(token)
|
| 208 |
+
|
| 209 |
+
# Get user UID and email from decoded token
|
|
|
|
| 210 |
uid = decoded.get('uid')
|
| 211 |
+
email = decoded.get('email') # Firebase ID tokens usually contain the email
|
| 212 |
+
|
| 213 |
if not uid:
|
| 214 |
+
print("Verified token does not contain a UID.")
|
| 215 |
return None
|
| 216 |
|
| 217 |
+
# Check if user is admin by querying Firebase Realtime Database using UID
|
|
|
|
| 218 |
admins_ref = db.reference("/admins")
|
| 219 |
admin_data = admins_ref.child(uid).get()
|
|
|
|
| 220 |
|
| 221 |
if admin_data and admin_data.get("is_admin", False):
|
| 222 |
+
# User is already an admin in the database
|
| 223 |
+
return decoded
|
| 224 |
+
elif email and email in ADMIN_EMAILS:
|
| 225 |
+
# User's email is in the approved list, but UID not found in DB.
|
| 226 |
+
# This is likely their first time accessing the API.
|
| 227 |
+
# Automatically create their admin entry using their UID.
|
| 228 |
+
print(f"First time admin access for {email} (UID: {uid}). Creating database entry.")
|
| 229 |
+
admins_ref.child(uid).set({
|
| 230 |
+
"email": email,
|
| 231 |
+
"is_admin": True,
|
| 232 |
+
"first_seen": datetime.datetime.utcnow().isoformat() + 'Z' # UTC Timestamp
|
| 233 |
+
})
|
| 234 |
+
print(f"Admin entry created for UID {uid}.")
|
| 235 |
+
# Return the decoded token as they are now an admin
|
| 236 |
return decoded
|
| 237 |
else:
|
| 238 |
+
# User is not in the approved admin list
|
| 239 |
+
print(f"User {uid} ({email}) is not in the approved admin list or not found in DB.")
|
| 240 |
return None
|
| 241 |
+
|
| 242 |
except firebase_auth.InvalidIdTokenError as e:
|
| 243 |
+
print(f"Invalid Firebase ID token provided: {e}")
|
| 244 |
return None
|
| 245 |
except firebase_auth.ExpiredIdTokenError as e:
|
| 246 |
+
print(f"Firebase ID token has expired: {e}")
|
| 247 |
return None
|
| 248 |
except firebase_auth.RevokedIdTokenError as e:
|
| 249 |
+
print(f"Firebase ID token has been revoked: {e}")
|
| 250 |
return None
|
| 251 |
except Exception as e:
|
| 252 |
+
print(f"Unexpected error during token verification: {e}")
|
| 253 |
+
# Consider logging the full traceback in production
|
| 254 |
+
# import traceback
|
| 255 |
+
# print(traceback.format_exc())
|
| 256 |
return None
|
| 257 |
|
| 258 |
|
| 259 |
+
|
| 260 |
# === Admin Setup (Legacy - kept for compatibility) ===
|
| 261 |
def setup_admins():
|
| 262 |
ref = db.reference("admins")
|