Expanded user coverage (#22)

Author: webbpinner

Makefile

@@ -91,7 +91,7 @@ release:          ## Create a new tag for release.
 docs:             ## Build the documentation.
 	@echo "building documentation ..."
 	@$(ENV_PREFIX)mkdocs build
-	URL="site/index.html"; xdg-open $$URL || sensible-browser $$URL || x-www-browser $$URL || gnome-open $$URL
+	URL="site/index.html"; xdg-open $$URL || sensible-browser $$URL || x-www-browser $$URL || gnome-open $$URL  || open $$URL
 
 .PHONY: switch-to-poetry
 switch-to-poetry: ## Switch to poetry package manager.

project_name/default.toml

@@ -5,6 +5,7 @@
 # SECRET_KEY = ""
 ALGORITHM = "HS256"
 ACCESS_TOKEN_EXPIRE_MINUTES = 30
+REFRESH_TOKEN_EXPIRE_MINUTES = 600
 
 [default.server]
 port = 8000

project_name/routes/__init__.py

@@ -1,14 +1,16 @@
 from fastapi import APIRouter
 
 from .content import router as content_router
+from .profile import router as profile_router
 from .security import router as security_router
 from .user import router as user_router
 
 main_router = APIRouter()
 
-main_router.include_router(user_router, prefix="/user", tags=["user"])
 main_router.include_router(content_router, prefix="/content", tags=["content"])
+main_router.include_router(profile_router, tags=["user"])
 main_router.include_router(security_router, tags=["security"])
+main_router.include_router(user_router, prefix="/user", tags=["user"])
 
 
 @main_router.get("/")

project_name/routes/profile.py

@@ -0,0 +1,10 @@
+from fastapi import APIRouter
+
+from ..security import AuthenticatedUser, User, UserResponse
+
+router = APIRouter()
+
+
+@router.get("/profile", response_model=UserResponse)
+async def my_profile(current_user: User = AuthenticatedUser):
+    return current_user

project_name/routes/security.py

@@ -5,15 +5,18 @@
 
 from ..config import settings
 from ..security import (
+    RefreshToken,
     Token,
     User,
     authenticate_user,
     create_access_token,
+    create_refresh_token,
     get_user,
+    validate_token,
 )
 
 ACCESS_TOKEN_EXPIRE_MINUTES = settings.security.access_token_expire_minutes
-
+REFRESH_TOKEN_EXPIRE_MINUTES = settings.security.refresh_token_expire_minutes
 
 router = APIRouter()
 
@@ -32,7 +35,39 @@ async def login_for_access_token(
 
     access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
     access_token = create_access_token(
-        data={"sub": user.username}, expires_delta=access_token_expires
+        data={"sub": user.username, "fresh": True},
+        expires_delta=access_token_expires,
+    )
+
+    refresh_token_expires = timedelta(minutes=REFRESH_TOKEN_EXPIRE_MINUTES)
+    refresh_token = create_refresh_token(
+        data={"sub": user.username}, expires_delta=refresh_token_expires
+    )
+
+    return {
+        "access_token": access_token,
+        "refresh_token": refresh_token,
+        "token_type": "bearer",
+    }
+
+
+@router.post("/refresh_token", response_model=Token)
+async def refresh_token(form_data: RefreshToken):
+    user = await validate_token(token=form_data.refresh_token)
+
+    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
+    access_token = create_access_token(
+        data={"sub": user.username, "fresh": False},
+        expires_delta=access_token_expires,
+    )
+
+    refresh_token_expires = timedelta(minutes=REFRESH_TOKEN_EXPIRE_MINUTES)
+    refresh_token = create_refresh_token(
+        data={"sub": user.username}, expires_delta=refresh_token_expires
     )
 
-    return {"access_token": access_token, "token_type": "bearer"}
+    return {
+        "access_token": access_token,
+        "refresh_token": refresh_token,
+        "token_type": "bearer",
+    }

project_name/routes/user.py

@@ -7,6 +7,7 @@
 from ..db import ActiveSession
 from ..security import (
     AdminUser,
+    AuthenticatedFreshUser,
     AuthenticatedUser,
     HashedPassword,
     User,
@@ -27,6 +28,15 @@ async def list_users(*, session: Session = ActiveSession):
 
 @router.post("/", response_model=UserResponse, dependencies=[AdminUser])
 async def create_user(*, session: Session = ActiveSession, user: UserCreate):
+
+    # verify user with username doesn't already exist
+    try:
+        await query_user(session=session, user_id_or_username=user.username)
+    except HTTPException:
+        pass
+    else:
+        raise HTTPException(status_code=422, detail="Username already exists")
+
     db_user = User.from_orm(user)
     session.add(db_user)
     session.commit()
@@ -37,7 +47,7 @@ async def create_user(*, session: Session = ActiveSession, user: UserCreate):
 @router.patch(
     "/{user_id}/password/",
     response_model=UserResponse,
-    dependencies=[AuthenticatedUser],
+    dependencies=[AuthenticatedFreshUser],
 )
 async def update_user_password(
     *,
@@ -85,16 +95,11 @@ async def query_user(
         )
     )
 
-    if not user:
+    if not user.first():
         raise HTTPException(status_code=404, detail="User not found")
     return user.first()
 
 
-@router.get("/me/", response_model=UserResponse)
-async def my_profile(current_user: User = AuthenticatedUser):
-    return current_user
-
-
 @router.delete("/{user_id}/", dependencies=[AdminUser])
 def delete_user(
     *, session: Session = ActiveSession, request: Request, user_id: int

project_name/security.py

@@ -22,9 +22,14 @@
 
 class Token(BaseModel):
     access_token: str
+    refresh_token: str
     token_type: str
 
 
+class RefreshToken(BaseModel):
+    refresh_token: str
+
+
 class TokenData(BaseModel):
     username: Optional[str] = None
 
@@ -116,7 +121,20 @@ def create_access_token(
         expire = datetime.utcnow() + expires_delta
     else:
         expire = datetime.utcnow() + timedelta(minutes=15)
-    to_encode.update({"exp": expire})
+    to_encode.update({"exp": expire, "scope": "access_token"})
+    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
+    return encoded_jwt
+
+
+def create_refresh_token(
+    data: dict, expires_delta: Optional[timedelta] = None
+) -> str:
+    to_encode = data.copy()
+    if expires_delta:
+        expire = datetime.utcnow() + expires_delta
+    else:
+        expire = datetime.utcnow() + timedelta(minutes=15)
+    to_encode.update({"exp": expire, "scope": "refresh_token"})
     encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
     return encoded_jwt
 
@@ -138,7 +156,7 @@ def get_user(username) -> Optional[User]:
 
 
 def get_current_user(
-    token: str = Depends(oauth2_scheme), request: Request = None
+    token: str = Depends(oauth2_scheme), request: Request = None, fresh=False
 ) -> User:
     credentials_exception = HTTPException(
         status_code=status.HTTP_401_UNAUTHORIZED,
@@ -156,6 +174,7 @@ def get_current_user(
     try:
         payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
         username: str = payload.get("sub")
+
         if username is None:
             raise credentials_exception
         token_data = TokenData(username=username)
@@ -164,6 +183,9 @@ def get_current_user(
     user = get_user(username=token_data.username)
     if user is None:
         raise credentials_exception
+    if fresh and (not payload["fresh"] and not user.superuser):
+        raise credentials_exception
+
     return user
 
 
@@ -178,6 +200,15 @@ async def get_current_active_user(
 AuthenticatedUser = Depends(get_current_active_user)
 
 
+def get_current_fresh_user(
+    token: str = Depends(oauth2_scheme), request: Request = None
+) -> User:
+    return get_current_user(token, request, True)
+
+
+AuthenticatedFreshUser = Depends(get_current_fresh_user)
+
+
 async def get_current_admin_user(
     current_user: User = Depends(get_current_user),
 ) -> User:
@@ -189,3 +220,9 @@ async def get_current_admin_user(
 
 
 AdminUser = Depends(get_current_admin_user)
+
+
+async def validate_token(token: str = Depends(oauth2_scheme)) -> User:
+
+    user = get_current_user(token=token)
+    return user

tests/test_profile.py

@@ -0,0 +1,10 @@
+def test_profile(api_client_authenticated):
+    response = api_client_authenticated.get("/profile")
+    assert response.status_code == 200
+    result = response.json()
+    assert "admin" in result["username"]
+
+
+def test_profile_no_auth(api_client):
+    response = api_client.get("/profile")
+    assert response.status_code == 401