User sessions.

Author: Temmmmmo

auth_backend/auth_plugins/auth_method.py

@@ -14,12 +14,13 @@
 
 from auth_backend.base import Base, ResponseModel
 from auth_backend.exceptions import AlreadyExists
-from auth_backend.models.db import AuthMethod, User, UserSession, Scope, UserSessionScope
+from auth_backend.models.db import AuthMethod, Scope, User, UserSession, UserSessionScope
 from auth_backend.schemas.types.scopes import Scope as TypeScope
 from auth_backend.settings import get_settings
 from auth_backend.utils.security import UnionAuth
 from auth_backend.utils.user_session_control import create_session
 
+
 logger = logging.getLogger(__name__)
 settings = get_settings()
 

auth_backend/auth_plugins/email.py

@@ -9,6 +9,7 @@
 from auth_backend.base import Base, ResponseModel
 from auth_backend.exceptions import AlreadyExists, AuthFailed, IncorrectUserAuthType, SessionExpired
 from auth_backend.models.db import AuthMethod, User, UserSession
+from auth_backend.schemas.types.scopes import Scope
 from auth_backend.settings import get_settings
 from auth_backend.utils.security import UnionAuth
 from auth_backend.utils.smtp import (
@@ -17,8 +18,9 @@
     send_confirmation_email,
     send_reset_email,
 )
-from .auth_method import AuthMethodMeta, Session, random_string, MethodMeta
-from auth_backend.schemas.types.scopes import Scope
+
+from .auth_method import AuthMethodMeta, MethodMeta, Session, random_string
+
 
 settings = get_settings()
 

auth_backend/models/db.py

@@ -4,13 +4,14 @@
 from typing import Iterator
 
 import sqlalchemy.orm
-from sqlalchemy import String, Integer, ForeignKey, DateTime, Boolean, func
+from sqlalchemy import Boolean, DateTime, ForeignKey, Integer, String, func
 from sqlalchemy.ext.hybrid import hybrid_property
-from sqlalchemy.orm import Mapped, mapped_column, relationship, backref, Session
+from sqlalchemy.orm import Mapped, Session, backref, mapped_column, relationship
 
 from auth_backend.exceptions import ObjectNotFound
 from auth_backend.settings import get_settings
 
+
 settings = get_settings()
 
 

auth_backend/routes/scopes.py

@@ -1,11 +1,11 @@
-from fastapi import APIRouter, HTTPException, Depends
+from fastapi import APIRouter, Depends, HTTPException
 from fastapi_sqlalchemy import db
 from pydantic import parse_obj_as
 from sqlalchemy import func
 
 from auth_backend.base import ResponseModel
-from auth_backend.models.db import UserSession, Scope
-from auth_backend.schemas.models import ScopeGet, ScopePost, ScopePatch
+from auth_backend.models.db import Scope, UserSession
+from auth_backend.schemas.models import ScopeGet, ScopePatch, ScopePost
 from auth_backend.utils.security import UnionAuth
 
 

auth_backend/routes/user_session.py

@@ -1,27 +1,29 @@
 import logging
+import string
 from datetime import datetime
 from typing import Literal
-import string
-from fastapi import APIRouter, Query, Depends
+
+from fastapi import APIRouter, Depends, Query
 from fastapi_sqlalchemy import db
-from starlette.responses import JSONResponse
 from sqlalchemy import not_
+from starlette.responses import JSONResponse
+
 from auth_backend.base import ResponseModel
-from auth_backend.exceptions import SessionExpired, ObjectNotFound
-from auth_backend.schemas.models import Session
-from auth_backend.models.db import AuthMethod, UserSession, User
+from auth_backend.exceptions import ObjectNotFound, SessionExpired
+from auth_backend.models.db import AuthMethod, User, UserSession
 from auth_backend.schemas.models import (
+    Session,
+    SessionPost,
+    SessionScopes,
     UserAuthMethods,
+    UserGet,
     UserGroups,
     UserIndirectGroups,
     UserInfo,
-    UserGet,
     UserScopes,
-    SessionScopes,
-    SessionPost,
 )
-from auth_backend.utils.security import UnionAuth
 from auth_backend.utils import user_session_control
+from auth_backend.utils.security import UnionAuth
 
 
 user_session = APIRouter(prefix="", tags=["User session"])
@@ -83,7 +85,9 @@ async def me(
 async def create_session(
     new_session: SessionPost, session: UserSession = Depends(UnionAuth(scopes=[], allow_none=False, auto_error=True))
 ):
-    return await user_session_control.create_session(session.user, new_session.scopes, db_session=db.session)
+    return await user_session_control.create_session(
+        session.user, new_session.scopes, new_session.expires, db_session=db.session
+    )
 
 
 @user_session.delete("/session/{token}")
@@ -108,13 +112,14 @@ async def delete_sessions(
     delete_current: bool = Query(default=False),
     current_session: UserSession = Depends(UnionAuth(scopes=[], allow_none=False, auto_error=True)),
 ):
-    other_sessions = current_session.user.active_sessions
-    for session in other_sessions:
-        if session.token == current_session.token and not delete_current:
-            continue
-        if session.expired:
-            raise SessionExpired(session.token)
-        session.expires = datetime.utcnow()
+    query = (
+        db.session.query(UserSession)
+        .filter(UserSession.user_id == current_session.user_id)
+        .filter(not_(UserSession.expired))
+    )
+    if not delete_current:
+        query = query.filter(UserSession.token != current_session.token)
+    query.update({"expires": datetime.utcnow()})
     db.session.commit()
 
 

auth_backend/schemas/models.py

@@ -1,11 +1,11 @@
 from __future__ import annotations
 
-from pydantic import Field
+from datetime import datetime
+
+from pydantic import Field, constr, validator
 
 from auth_backend.base import Base
 from auth_backend.schemas.types.scopes import Scope
-from pydantic import constr
-from datetime import datetime
 
 
 class PinchedScope(Base):
@@ -136,6 +136,12 @@ class SessionPost(Base):
     scopes: list[Scope] = []
     expires: datetime | None = None
 
+    @validator("expires")
+    def expires_validator(self, exp):
+        if exp < datetime.utcnow():
+            raise ValueError("Session is expired")
+        return exp
+
 
 Group.update_forward_refs()
 GroupGet.update_forward_refs()

auth_backend/utils/user_session_control.py

@@ -1,16 +1,17 @@
+import random
+import string
 from datetime import datetime, timedelta
 
-from auth_backend.models.db import User, UserSession, Scope, UserSessionScope
-from auth_backend.schemas.models import Session
-from auth_backend.schemas.types.scopes import Scope as TypeScope
-from sqlalchemy.orm import Session as DbSession
-from fastapi_sqlalchemy import db
 from fastapi import HTTPException
+from fastapi_sqlalchemy import db
+from sqlalchemy.orm import Session as DbSession
+
 from auth_backend.base import ResponseModel
-from auth_backend.settings import Settings
-import random
-import string
-from auth_backend.settings import get_settings
+from auth_backend.models.db import Scope, User, UserSession, UserSessionScope
+from auth_backend.schemas.models import Session
+from auth_backend.schemas.types.scopes import Scope as TypeScope
+from auth_backend.settings import Settings, get_settings
+
 
 settings = get_settings()
 

migrations/env.py

@@ -1,12 +1,12 @@
 from logging.config import fileConfig
 
-from sqlalchemy import engine_from_config
-from sqlalchemy import pool
-
 from alembic import context
+from sqlalchemy import engine_from_config, pool
+
 from auth_backend.models.base import Base
 from auth_backend.settings import get_settings
 
+
 # this is the Alembic Config object, which provides
 # access to the values within the .ini file in use.
 config = context.config

tests/conftest.py

@@ -6,11 +6,11 @@
 from sqlalchemy import create_engine
 from sqlalchemy.orm import sessionmaker
 from starlette import status
+
 from auth_backend.auth_plugins.auth_method import random_string
-from auth_backend.models.db import Group, UserGroup, User, GroupScope
+from auth_backend.models.db import AuthMethod, Group, GroupScope, Scope, User, UserGroup, UserSession, UserSessionScope
 from auth_backend.routes.base import app
 from auth_backend.settings import get_settings
-from auth_backend.models.db import AuthMethod, UserSession, Scope, UserSessionScope
 
 
 @pytest.fixture

tests/test_routes/test_user_sessions.py

@@ -1,12 +1,13 @@
 import logging
+from datetime import datetime, timedelta
 from time import sleep
 
-from starlette.testclient import TestClient
 from sqlalchemy.orm import Session
-from datetime import datetime
 from starlette import status
-from auth_backend.models.db import UserSession, UserSessionScope, GroupScope, Group, UserGroup
-from auth_backend.models.db import Scope
+from starlette.testclient import TestClient
+
+from auth_backend.models.db import Group, GroupScope, Scope, UserGroup, UserSession, UserSessionScope
+
 
 logger = logging.getLogger(__name__)
 
@@ -52,6 +53,11 @@ def test_create_session(client_auth: TestClient, dbsession: Session, user_scopes
     new_session_: UserSession = dbsession.query(UserSession).get(new_session2.json()["id"])
     assert new_session_.id != current_session_id_get.json()["id"]
     assert new_session_.scopes != current_session_scopes_get.json()["session_scopes"]
+    time = datetime.utcnow() + timedelta(days=999999)
+    new_session3 = client_auth.post("/session", headers=header, json={"expires": str(time)})
+    assert new_session3.status_code == status.HTTP_200_OK
+    new_session3_: UserSession = dbsession.query(UserSession).get(new_session3.json()['id'])
+    assert new_session3_.expires == time
     dbsession.query(UserSessionScope).filter(
         UserSessionScope.scope_id == scope1.id, UserSessionScope.scope_id == scope2.id
     ).delete()
@@ -60,6 +66,7 @@ def test_create_session(client_auth: TestClient, dbsession: Session, user_scopes
         UserSession.user_id == user_id,
         UserSession.id == new_session1.json()["id"],
         UserSession.id == new_session2.json()["id"],
+        UserSession.id == new_session3.json()["id"],
     ).delete()
     dbsession.query(UserGroup).filter(UserGroup.group_id == group_id).delete()
     dbsession.delete(scope1)
@@ -159,3 +166,6 @@ def test_get_sessions(client_auth: TestClient, dbsession: Session, user_scopes):
     assert current_session.token in list(all_sessions.json()[i]['token'] for i in range(len(all_sessions.json())))
     assert new_session1.token in list(all_sessions.json()[i]['token'] for i in range(len(all_sessions.json())))
     assert new_session2.token in list(all_sessions.json()[i]['token'] for i in range(len(all_sessions.json())))
+    client_auth.delete(f'/session/{new_session1.token}', headers=header)
+    all_sessions = client_auth.get("/session", headers=header)
+    assert new_session1.token not in list(all_sessions.json()[i]['token'] for i in range(len(all_sessions.json())))