feat: sua expiry handling

src/iam/router.py

@@ -75,7 +75,7 @@ from src.iam.schemas import (
 	IAMPutGroupInvitationRequest,
 	IAMPutGroupInvitationAcceptRequest,
 )
-from src.utils import decode_jwt
+from src.utils import verify_email_token
 
 router = APIRouter(
 	tags=["IAM"],
@@ -373,11 +373,9 @@ async def accept_invitation(
 	user_model: user_model_claims_dependency,
 	request_model: IAMPutGroupInvitationAcceptRequest,
 ):
-	email_claims = await decode_jwt(request_model.jwt)
-	claimed_email = email_claims["email"]
-
-	if user_model.email != claimed_email:
-		raise UnauthorizedException("The logged in user and email do not match.")
+	email_claims = await verify_email_token(
+		token=request_model.jwt, user_model=user_model
+	)
 
 	org_model = db.get(Org, email_claims["org_id"])
 	if org_model is None:

src/user/router.py

@@ -10,7 +10,6 @@ Endpoints:
 
 from fastapi import APIRouter, status, BackgroundTasks
 
-from src.auth.exceptions import UnauthorizedException
 from src.organisation.exceptions import OrgNotFoundException
 from src.user.schemas import (
 	UserResponse,
@@ -32,7 +31,7 @@ from src.auth.dependencies import (
 )
 from src.auth.service import claims_dependency
 from src.database import db_dependency
-from src.utils import decode_jwt
+from src.utils import verify_email_token
 
 router = APIRouter(
 	prefix="/user",
@@ -181,11 +180,9 @@ async def accept_invitation(
 	user_model: user_model_claims_dependency,
 	request_model: UserPostInvitationAcceptRequest,
 ):
-	email_claims = await decode_jwt(request_model.jwt)
-	claimed_email = email_claims["email"]
-
-	if user_model.email != claimed_email:
-		raise UnauthorizedException("The logged in user and email do not match.")
+	email_claims = await verify_email_token(
+		token=request_model.jwt, user_model=user_model
+	)
 
 	org_model = db.get(Org, email_claims["org_id"])
 	if org_model is None:

src/utils.py

@@ -1,3 +1,4 @@
+from datetime import datetime, timezone
 from joserfc import jwt, jwk, errors
 
 from src.auth.exceptions import UnauthorizedException
@@ -21,6 +22,22 @@ async def decode_jwt(encoded):
 		raise UnauthorizedException("Invalid JWS")
 
 
+async def verify_email_token(user_model, token):
+	email_claims = await decode_jwt(token)
+
+	claimed_email = email_claims["email"]
+
+	expiry = datetime.fromtimestamp(email_claims["exp"], timezone.utc)
+
+	if expiry < datetime.now(timezone.utc):
+		raise UnauthorizedException("Invitation expired.")
+
+	if user_model.email != claimed_email:
+		raise UnauthorizedException("The logged in user and email do not match.")
+
+	return email_claims
+
+
 async def send_email(recipient: str, subject: str, body: str):
 	print(recipient)
 	print(subject)