add fingerprint httpbearer schema

src/api/dependacies/user_dep.py

@@ -1,7 +1,11 @@
 from typing import Annotated
 
 from fastapi import Depends, HTTPException, Path
-from fastapi.security import HTTPBearer, OAuth2PasswordBearer
+from fastapi.security import (
+    HTTPAuthorizationCredentials,
+    HTTPBearer,
+    OAuth2PasswordBearer,
+)
 from jwt import InvalidTokenError
 
 from src.api.dependacies.db_dep import sessionDep
@@ -12,8 +16,10 @@ from src.services.tasks import TaskService
 from src.services.users import UserService
 
 http_bearer = HTTPBearer(auto_error=False)
+
 oauth2_scheme = OAuth2PasswordBearer(tokenUrl=f"{settings.api.v1_login_url}/login")
-AccessTokenDep = Annotated[str, Depends(oauth2_scheme)]
+# AccessTokenDep = Annotated[str, Depends(oauth2_scheme)]
+AccessTokenDep = Annotated[HTTPAuthorizationCredentials, Depends(http_bearer)]
 
 
 async def get_current_user(
@@ -25,7 +31,7 @@ async def get_current_user(
         headers={"WWW-Authenticate": "Bearer"},
     )
     try:
-        payload = AuthManager.decode_access_token(token, verify_exp)
+        payload = AuthManager.decode_access_token(token.credentials, verify_exp)
         if payload is None:
             raise credentials_exception
         user = TokenData(**payload)

src/api/v1/auth.py

@@ -1,6 +1,6 @@
 from typing import Annotated
 
-from fastapi import APIRouter, Cookie, Depends, HTTPException, Response
+from fastapi import APIRouter, Body, Cookie, Depends, Form, HTTPException, Response
 from fastapi.security import OAuth2PasswordRequestForm
 
 from src.api.dependacies.db_dep import sessionDep
@@ -27,15 +27,16 @@ async def login(
     session: sessionDep,
     credential: Annotated[OAuth2PasswordRequestForm, Depends()],
     response: Response,
+    fingerprint: str = Form(),
 ):
-    result = await AuthService(session).login(credential.username, credential.password)
+    result = await AuthService(session).login(credential.username, credential.password, fingerprint=fingerprint)
     response.set_cookie(
         key="refresh_token",
         value=result["refresh_token"],
         httponly=True,
         samesite="lax",
         path=settings.api.v1.auth,
-        max_age=60 * 60 * 24 * 7,
+        max_age=60 * 60 * 24 * settings.refresh_token.expire_days,
     )
     return result
 
@@ -43,20 +44,21 @@ async def login(
 @router.post(path="/refresh", response_model=Token)
 async def refresh(
     session: sessionDep,
-    response: Response,
     current_user: RefreshUser,
+    response: Response,
+    fingerprint: str = Body(),
     refresh_token: Annotated[str | None, Cookie(name="refresh_token")] = None,
 ):
     if refresh_token is None:
         raise HTTPException(status_code=401, detail="No refresh token")
-    result = await AuthService(session).refresh_tokens(refresh_token, current_user)
+    result = await AuthService(session).refresh_tokens(refresh_token, current_user, fingerprint)
     response.set_cookie(
         key="refresh_token",
         value=result["refresh_token"],
         httponly=True,
         samesite="lax",
         path=settings.api.v1.auth,
-        max_age=60 * 60 * 24 * 7,
+        max_age=60 * 60 * 24 * settings.refresh_token.expire_days,
     )
     return result
 

src/core/settings.py

@@ -30,20 +30,23 @@ class DbSettings(BaseModel):
     url: str = f"sqlite+aiosqlite:///{DB_PATH}"
 
 
-class AccessToken(BaseSettings):
+class AccessToken(BaseModel):
-    model_config = SettingsConfigDict(
-        env_file=".env", env_file_encoding="utf-8", env_prefix="ACCESS_TOKEN_"
-    )
     expire_minutes: int
     secret_key: str
     algorithm: str = "HS256"
     token_type: str = "bearer"  # noqa: S105
 
 
+class RefreshToken(BaseModel):
+    expire_days: int
+
+
 class Settings(BaseSettings):
     api: ApiPrefix = ApiPrefix()
     db: DbSettings = DbSettings()
-    access_token: AccessToken = AccessToken()  # type: ignore
+    access_token: AccessToken  
+    refresh_token: RefreshToken
+    model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8", env_nested_delimiter='__')
 
 
-settings = Settings()
+settings = Settings()  # type: ignore

src/migrations/versions/2025_09_21_2016-5821f37941a8_expire_and_fingerprintjs_token.py

@@ -0,0 +1,33 @@
+"""expire and fingerprintjs token
+
+Revision ID: 5821f37941a8
+Revises: b879d3502c37
+Create Date: 2025-09-21 20:16:48.289050
+
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = '5821f37941a8'
+down_revision: Union[str, None] = 'b879d3502c37'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column('refresh_tokens', sa.Column('fingerprint', sa.String(length=255), nullable=False))
+    op.add_column('refresh_tokens', sa.Column('expired_at', sa.TIMESTAMP(timezone=True), server_default=sa.text("(datetime('now', '+7 days'))"), nullable=False))
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_column('refresh_tokens', 'expired_at')
+    op.drop_column('refresh_tokens', 'fingerprint')
+    # ### end Alembic commands ###

src/models/tokens.py

@@ -1,7 +1,10 @@
-from sqlalchemy import ForeignKey, Integer, String
+from datetime import datetime
+
+from sqlalchemy import TIMESTAMP, ForeignKey, Integer, String, text
 from sqlalchemy.orm import Mapped, mapped_column
 
 from src.core.database import Base
+from src.core.settings import settings
 
 
 class RefreshTokensORM(Base):
@@ -9,3 +12,8 @@ class RefreshTokensORM(Base):
     id: Mapped[int] = mapped_column(Integer, primary_key=True)
     token: Mapped[str] = mapped_column(String(255), nullable=False)
     user_id: Mapped[int] = mapped_column(Integer, ForeignKey("users.id"))
+    fingerprint: Mapped[str] = mapped_column(String(255), nullable=False)
+    expired_at: Mapped[datetime] = mapped_column(
+        TIMESTAMP(timezone=True),
+        server_default=text(f"datetime('now', '+{settings.refresh_token.expire_days} days')"),
+    )

src/services/auth.py

@@ -19,7 +19,7 @@ class AuthService(BaseService):
         await self.session.commit()
         return User.model_validate(result)
 
-    async def login(self, username: str, password: str):
+    async def login(self, username: str, password: str, fingerprint: str) -> dict:
         result = await self.session.user.get_one_or_none(username=username)
         if result is None:
             raise HTTPException(
@@ -27,7 +27,7 @@ class AuthService(BaseService):
                 detail="Incorrect username or password",
             )
         user = UserWithHashedPass.model_validate(result)
-        user_token = TokenData.model_validate(user.model_dump())
+        token_data = TokenData.model_validate(user.model_dump())
         verify = AuthManager.verify_password(
             plain_password=password, hashed_password=user.hashed_password
         )
@@ -36,9 +36,13 @@ class AuthService(BaseService):
                 status_code=401,
                 detail="Incorrect username or password",
             )
-        access_token = AuthManager.create_access_token(user_token.model_dump())
+        access_token = AuthManager.create_access_token(token_data.model_dump())
         refresh_token = AuthManager.create_refresh_token()
-        await self.session.auth.create_one({"token": refresh_token, "user_id": user.id})
+        await self.session.auth.create_one({
+            "token": refresh_token, 
+            "user_id": user.id,
+            "fingerprint": fingerprint
+        })
         await self.session.commit()
         return {
             "access_token": access_token,
@@ -50,7 +54,7 @@ class AuthService(BaseService):
         await self.session.auth.delete_one(token=token)
         await self.session.commit()
 
-    async def refresh_tokens(self, refresh_token: str, user_data: TokenData):
+    async def refresh_tokens(self, refresh_token: str, user_data: TokenData, fingerprint: str) -> dict:
         token_record = await self.session.auth.get_one_or_none(token=refresh_token)
         if not token_record or token_record.user_id != user_data.id:
             raise HTTPException(status_code=401, detail="Invalid refresh token")
@@ -60,6 +64,7 @@ class AuthService(BaseService):
         await self.session.auth.create_one({
             "token": new_refresh_token,
             "user_id": user_data.id,
+            "fingerprint": fingerprint
         })
         await self.session.commit()
         return {