aboutsummaryrefslogtreecommitdiff
path: root/app/routes
diff options
context:
space:
mode:
Diffstat (limited to 'app/routes')
-rw-r--r--app/routes/auth_routes.py (renamed from app/routes/token_route.py)43
-rw-r--r--app/routes/refresh_route.py32
-rw-r--r--app/routes/user_routes.py127
3 files changed, 100 insertions, 102 deletions
diff --git a/app/routes/token_route.py b/app/routes/auth_routes.py
index 124e0cc..a28ec63 100644
--- a/app/routes/token_route.py
+++ b/app/routes/auth_routes.py
@@ -1,22 +1,22 @@
-from fastapi import APIRouter, status, Depends, HTTPException
-from fastapi.responses import JSONResponse, Response
-from typing import Annotated
+from fastapi import Depends, APIRouter, status, HTTPException
+from fastapi.security import OAuth2PasswordRequestForm
+from fastapi.responses import Response
from datetime import timedelta
from typing import Annotated
-from fastapi.security import OAuth2PasswordRequestForm
-from app.util.db_dependency import get_db
from app.util.authentication import (
- authenticate_user,
create_access_token,
+ authenticate_user,
+ refresh_get_current_user,
)
-from app.schemas.auth_schemas import Token
+from app.schemas.auth_schemas import Token, User
+from app.util.db_dependency import get_db
-router = APIRouter(prefix="/token", tags=["token"])
+router = APIRouter(prefix="/auth", tags=["auth"])
-@router.post("/")
+@router.post("/token", summary="Authenticate and get an access token")
async def login_for_access_token(
form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
response: Response,
@@ -26,6 +26,7 @@ async def login_for_access_token(
Return an access token for the user, if the given authentication details are correct
"""
user = authenticate_user(db, form_data.username, form_data.password)
+ print(user)
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
@@ -34,14 +35,14 @@ async def login_for_access_token(
)
access_token_expires = timedelta(minutes=15)
access_token = create_access_token(
- data={"sub": user.username, "refresh": False},
+ data={"sub": user.id, "refresh": False},
expires_delta=access_token_expires,
)
# Create a refresh token - just an access token with a longer expiry
# and more restrictions ("refresh" is True)
refresh_token_expires = timedelta(days=1)
refresh_token = create_access_token(
- data={"sub": user.username, "refresh": True},
+ data={"sub": user.id, "refresh": True},
expires_delta=refresh_token_expires,
)
# response = JSONResponse(content={"success": True})
@@ -58,3 +59,23 @@ async def login_for_access_token(
refresh_token=refresh_token,
token_type="bearer",
)
+
+
+# Full native JWT support is not complete in FastAPI yet :(
+# Part of that is token refresh, so we must implement it ourselves
+@router.post("/refresh")
+async def refresh_access_token(
+ current_user: Annotated[User, Depends(refresh_get_current_user)],
+) -> Token:
+ """
+ Return a new access token if the refresh token is valid
+ """
+ access_token_expires = timedelta(minutes=30)
+ access_token = create_access_token(
+ data={"sub": current_user.id, "refresh": False},
+ expires_delta=access_token_expires,
+ )
+ return Token(
+ access_token=access_token,
+ token_type="bearer",
+ )
diff --git a/app/routes/refresh_route.py b/app/routes/refresh_route.py
deleted file mode 100644
index a8b7a5c..0000000
--- a/app/routes/refresh_route.py
+++ /dev/null
@@ -1,32 +0,0 @@
-from fastapi import Depends, APIRouter
-from datetime import timedelta
-from typing import Annotated
-
-from app.util.authentication import (
- create_access_token,
- refresh_get_current_user,
-)
-from app.schemas.auth_schemas import Token, User
-
-
-router = APIRouter(prefix="/refresh", tags=["refresh"])
-
-
-# Full native JWT support is not complete in FastAPI yet :(
-# Part of that is token refresh, so we must implement it ourselves
-@router.post("/")
-async def refresh_access_token(
- current_user: Annotated[User, Depends(refresh_get_current_user)],
-) -> Token:
- """
- Return a new access token if the refresh token is valid
- """
- access_token_expires = timedelta(minutes=30)
- access_token = create_access_token(
- data={"sub": current_user.username, "refresh": False},
- expires_delta=access_token_expires,
- )
- return Token(
- access_token=access_token,
- token_type="bearer",
- )
diff --git a/app/routes/user_routes.py b/app/routes/user_routes.py
index 53c8943..1c6c61e 100644
--- a/app/routes/user_routes.py
+++ b/app/routes/user_routes.py
@@ -14,72 +14,23 @@ from models import User as UserModel
from app.util.authentication import get_current_user_from_token
-router = APIRouter(prefix="/user", tags=["user"])
+router = APIRouter(prefix="/users", tags=["users"])
-# In order to help protect some anonymity/privacy, user routes
-# do not use path parameters, as then people could potentially
-# see if a specific username exists or not. Instead, the user
-# routes will use query parameters to specify the user to act
-
-@router.post("/register", summary="Register a new user")
-async def get_links(
- login_data: LoginDataSchema,
- db=Depends(get_db),
-):
- """
- Given the login data (username, password) process the registration of a new
- user account and return either the user or an error message
- """
- username = login_data.username
- password = login_data.password
- # Make sure the password meets all of the requirements
- # if len(password) < 8:
- # raise HTTPException(
- # status_code=status.HTTP_400_BAD_REQUEST,
- # detail="Password must be at least 8 characters",
- # )
- # if not any(char.isdigit() for char in password):
- # raise HTTPException(
- # status_code=status.HTTP_400_BAD_REQUEST,
- # detail="Password must contain at least one digit",
- # )
- # if not any(char.isupper() for char in password):
- # raise HTTPException(
- # status_code=status.HTTP_400_BAD_REQUEST,
- # detail="Password must contain at least one uppercase letter",
- # )
- # Make sure the username isn't taken
- user = db.query(UserModel).filter(UserModel.username == username).first()
- if user:
- raise HTTPException(
- status_code=status.HTTP_409_CONFLICT,
- detail="Username not available",
- )
- # Otherwise, hash the password, create the api key, and add the new user
- hashed_password = bcrypt.hashpw(
- password.encode("utf-8"), bcrypt.gensalt()
- ).decode("utf-8")
- api_key = "".join(
- random.choices(string.ascii_letters + string.digits, k=20)
- )
- new_user = UserModel(
- username=username, hashed_password=hashed_password, api_key=api_key
- )
- db.add(new_user)
- db.commit()
-
- return status.HTTP_201_CREATED
-
-
-@router.get("/delete", summary="Delete a user - provided it's your own")
+@router.delete("/{user_id}", summary="Delete your account")
async def delete_user(
+ user_id: Annotated[int, Path(title="Link to delete")],
current_user: Annotated[User, Depends(get_current_user_from_token)],
db=Depends(get_db),
):
"""
Delete the user account associated with the current user
"""
+ if user_id != current_user.id:
+ raise HTTPException(
+ status_code=status.HTTP_403_FORBIDDEN,
+ detail="You can only delete your own account",
+ )
user = db.query(UserModel).filter(UserModel.id == current_user.id).first()
if not user:
raise HTTPException(
@@ -91,15 +42,21 @@ async def delete_user(
return status.HTTP_204_NO_CONTENT
-@router.put("/updatepass", summary="Update your account's password")
+@router.post("/{user_id}", summary="Update your account password")
async def update_pass(
- current_user: Annotated[User, Depends(get_current_user_from_token)],
+ user_id: Annotated[int, Path(title="Link to update")],
update_data: UpdatePasswordSchema,
+ current_user: Annotated[User, Depends(get_current_user_from_token)],
db=Depends(get_db),
):
"""
Update the pass of the current user account
"""
+ if user_id != current_user.id:
+ raise HTTPException(
+ status_code=status.HTTP_403_FORBIDDEN,
+ detail="You can only update your own account",
+ )
# Make sure the password meets all of the requirements
# if len(update_data.new_password) < 8:
# raise HTTPException(
@@ -128,3 +85,55 @@ async def update_pass(
).decode("utf-8")
db.commit()
return status.HTTP_204_NO_CONTENT
+
+
+@router.post("/register", summary="Register a new user")
+async def get_links(
+ login_data: LoginDataSchema,
+ db=Depends(get_db),
+):
+ """
+ Given the login data (username, password) process the registration of a new
+ user account and return either the user or an error message
+ """
+ username = login_data.username
+ password = login_data.password
+ print(username)
+ print(password)
+ # Make sure the password meets all of the requirements
+ # if len(password) < 8:
+ # raise HTTPException(
+ # status_code=status.HTTP_400_BAD_REQUEST,
+ # detail="Password must be at least 8 characters",
+ # )
+ # if not any(char.isdigit() for char in password):
+ # raise HTTPException(
+ # status_code=status.HTTP_400_BAD_REQUEST,
+ # detail="Password must contain at least one digit",
+ # )
+ # if not any(char.isupper() for char in password):
+ # raise HTTPException(
+ # status_code=status.HTTP_400_BAD_REQUEST,
+ # detail="Password must contain at least one uppercase letter",
+ # )
+ # Make sure the username isn't taken
+ user = db.query(UserModel).filter(UserModel.username == username).first()
+ if user:
+ raise HTTPException(
+ status_code=status.HTTP_409_CONFLICT,
+ detail="Username not available",
+ )
+ # Otherwise, hash the password, create the api key, and add the new user
+ hashed_password = bcrypt.hashpw(
+ password.encode("utf-8"), bcrypt.gensalt()
+ ).decode("utf-8")
+ api_key = "".join(
+ random.choices(string.ascii_letters + string.digits, k=20)
+ )
+ new_user = UserModel(
+ username=username, hashed_password=hashed_password, api_key=api_key
+ )
+ db.add(new_user)
+ db.commit()
+
+ return status.HTTP_201_CREATED
generated by cgit v1.2.3-70-g09d2 (git 2.46.0) at 2025-07-18 12:53:25 +0000