users table endpoints. auth to fix

src/api/auth.py

@@ -0,0 +1,33 @@
+from datetime import timedelta
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, HTTPException, status
+from fastapi.security import OAuth2PasswordRequestForm
+from psycopg2._psycopg import connection
+
+from api.models import Token
+from api.utils import authenticate_user, create_access_token
+from db.internal import get_db_connection
+from settings import settings
+
+auth_router = APIRouter(prefix="/api", tags=["auth"])
+
+
+@auth_router.post("/token")
+async def login(
+    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
+    conn: Annotated[connection, Depends(get_db_connection)]
+) -> Token:
+    user = authenticate_user(conn, form_data.username, form_data.password)  # change db
+    if not user:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Incorrect username or password",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+
+    access_token_expire_time = timedelta(minutes=settings.access_token_expiration_time)
+    access_token = create_access_token(
+        data={"sub": user.username}, expires_delta=access_token_expire_time
+    )
+    return Token(access_token=access_token, token_type="bearer")

src/api/models.py

@@ -0,0 +1,28 @@
+from datetime import datetime
+
+from pydantic import BaseModel
+
+
+class Token(BaseModel):
+    access_token: str
+    token_type: str
+
+
+class TokenData(BaseModel):
+    username: str
+
+
+class User(BaseModel):
+    def fill(self, params):
+        self.username = params['username']
+        self.password = params['password']
+        self.disabled = params['disabled']
+        self.groups_ids = params['groups_ids']
+        self.last_seen_at = params['last_seen_at']
+        self.created_at = params['created_at']
+    username: str = ''
+    password: str = ''
+    disabled: bool = False
+    groups_ids: list[str] | None = None
+    last_seen_at: datetime | None = None
+    created_at: datetime | None = None

src/api/status.py

@@ -0,0 +1,8 @@
+from fastapi import APIRouter
+
+status_router = APIRouter(prefix="/api", tags=["status"])
+
+
+@status_router.get('/ping')
+async def ping():
+    return {'ok'}

src/api/tests.py

@@ -0,0 +1,21 @@
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, HTTPException, status
+
+from api.models import User
+from api.utils import get_current_user
+
+test_router = APIRouter(prefix="/api", tags=["test"])
+
+
+@test_router.get('/test-private')
+async def test_private_func(token: Annotated[User, Depends(get_current_user)]):
+    return {'private nya'}
+
+
+@test_router.post('/test')
+async def test_func(text: str):
+    print(text)
+    if text == 'thighs':
+        raise HTTPException(status_code=status.HTTP_402_PAYMENT_REQUIRED)
+    return {'nya'}

src/api/users.py

@@ -0,0 +1,33 @@
+from typing import Annotated
+
+from fastapi import APIRouter, Depends
+from psycopg2._psycopg import connection
+
+import db.users as db
+from api.models import User
+from api.utils import get_current_user
+from db.internal import get_db_connection
+
+users_router = APIRouter(prefix="/api/users", tags=["users"])
+
+
+@users_router.get("/me")
+async def read_users_me(current_user: Annotated[User, Depends(get_current_user)]):
+    return current_user
+
+
+@users_router.post("/user")
+async def read_users_any(username: str, conn: Annotated[connection, Depends(get_db_connection)]):
+    user = User()
+    user.fill(db.get_user(conn, username))
+    return user
+
+
+@users_router.post("/add")
+async def add_user(username: str, password: str, conn: Annotated[connection, Depends(get_db_connection)]):
+    return db.create_user(conn, username, password)
+
+
+@users_router.post("/delete")
+async def delete_user(username: str, conn: Annotated[connection, Depends(get_db_connection)]):
+    return db.delete_user(conn, username)

src/api/utils.py

@@ -0,0 +1,88 @@
+from datetime import datetime, timedelta, timezone
+from typing import Annotated
+
+import jwt
+from fastapi import Depends, HTTPException, status
+from fastapi.security import OAuth2PasswordBearer
+from jwt.exceptions import InvalidTokenError
+from passlib.context import CryptContext
+from psycopg2._psycopg import connection
+
+import db.users
+import settings.settings as settings
+from api.models import TokenData, User
+from db.internal import get_db_connection
+
+pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
+
+
+def decode_token(token):
+    return jwt.decode(token, settings.secret_key, algorithms=[settings.algorithm])
+
+def encode_token(payload):
+    return jwt.encode(payload, settings.secret_key, algorithm=settings.algorithm)
+
+def verify_password(plain_password, hashed_password):
+    return pwd_context.verify(plain_password, hashed_password)
+
+
+
+def authenticate_user(
+    conn: connection,
+    username: str,
+    password: str
+):
+    user = User()
+    userdata = db.users.get_user(conn, username)
+    if not userdata:
+        return False
+    if not verify_password(password, user.password):
+        return False
+    user.fill(userdata)
+    return user
+
+def create_access_token(
+    data: dict,
+    expires_delta: timedelta
+):
+    encode_payload = data.copy()
+    expire_moment = datetime.now(timezone.utc) + expires_delta
+    encode_payload.update({"exp": expire_moment})
+    encoded_jwt = encode_token(encode_payload)
+    return encoded_jwt
+
+
+async def get_current_user(
+    token: Annotated[str, Depends(oauth2_scheme)],
+    conn: Annotated[connection, Depends(get_db_connection)]
+):
+    credentials_exception = HTTPException(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        detail="Could not validate credentials",
+        headers={"WWW-Authenticate": "Bearer"},
+    )
+
+    try:
+        payload = decode_token(token)
+        print(payload)
+        username = payload.get("sub")
+        if username is None:
+            raise credentials_exception
+        token_data = TokenData(username=username)
+    except InvalidTokenError:
+        raise credentials_exception
+
+    user = User()
+    user.fill(db.users.get_user(conn, username=token_data.username))
+    if user is None:
+        raise credentials_exception
+
+    if user.disabled:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Inactive user"
+        )
+
+    return user