feat(fastapi): render a 500 html response when unique SID generation fails

We've seen a bug in the past where unique SID generation fails and
still ends up raising an exception.

This commit reworks how we deal with database exceptions internally,
tries for 36 iterations to set a fresh unique SID, and raises a 500
HTTPException if we were unable to.

Signed-off-by: Kevin Morris <kevr@0cost.org>
This commit is contained in:
Kevin Morris 2021-12-02 23:26:42 -08:00
parent abfd41f31e
commit 806a19b91a
No known key found for this signature in database
GPG key ID: F7E46DED420788F3
3 changed files with 85 additions and 29 deletions

View file

@ -1,12 +1,14 @@
import hashlib
from datetime import datetime
from http import HTTPStatus
from typing import List, Set
import bcrypt
from fastapi import Request
from fastapi import HTTPException, Request
from sqlalchemy import or_
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import backref, relationship
import aurweb.config
@ -108,33 +110,45 @@ class User(Base):
if not self.authenticated:
return None
now_ts = datetime.utcnow().timestamp()
session_ts = now_ts + (
session_time if session_time
else aurweb.config.getint("options", "login_timeout")
)
# Maximum number of iterations where we attempt to generate
# a unique SID. In cases where the Session table has
# exhausted all possible values, this will catch exceptions
# instead of raising them and include details about failing
# generation in an HTTPException.
tries = 36
sid = None
exc = None
for i in range(tries):
exc = None
now_ts = datetime.utcnow().timestamp()
session_ts = now_ts + (
session_time if session_time
else aurweb.config.getint("options", "login_timeout")
)
try:
with db.begin():
self.LastLogin = now_ts
self.LastLoginIPAddress = request.client.host
if not self.session:
sid = generate_unique_sid()
self.session = db.create(Session, User=self,
SessionID=sid,
LastUpdateTS=session_ts)
else:
last_updated = self.session.LastUpdateTS
if last_updated and last_updated < now_ts:
self.session.SessionID = generate_unique_sid()
self.session.LastUpdateTS = session_ts
break
except IntegrityError as exc_:
exc = exc_
with db.begin():
self.LastLogin = now_ts
self.LastLoginIPAddress = request.client.host
if not self.session:
sid = generate_unique_sid()
self.session = Session(UsersID=self.ID, SessionID=sid,
LastUpdateTS=session_ts)
db.add(self.session)
else:
last_updated = self.session.LastUpdateTS
if last_updated and last_updated < now_ts:
self.session.SessionID = sid = generate_unique_sid()
else:
# Session is still valid; retrieve the current SID.
sid = self.session.SessionID
if exc:
detail = ("Unable to generate a unique session ID in "
f"{tries} iterations.")
raise HTTPException(status_code=HTTPStatus.INTERNAL_SERVER_ERROR,
detail=detail)
self.session.LastUpdateTS = session_ts
request.cookies["AURSID"] = self.session.SessionID
return self.session.SessionID
def has_credential(self, credential: Set[int],
@ -142,8 +156,7 @@ class User(Base):
from aurweb.auth.creds import has_credential
return has_credential(self, credential, approved)
def logout(self, request):
del request.cookies["AURSID"]
def logout(self, request: Request):
self.authenticated = False
if self.session:
with db.begin():