initialize chapter8 secured api

Author: giunio-prc

Chapter08/secured-api/.python-version

@@ -0,0 +1 @@
+3.13

Chapter08/secured-api/README.md

@@ -0,0 +1,45 @@
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
+from typing import Annotated, TypedDict
+
+from fastapi import Depends, FastAPI
+from security.authenticator import (
+    BaseAuthenticator,
+    UnsafeAuthenticator,
+)
+
+from security.commons import UserInfo
+from security.dependencies import GetUserWithRole, get_user
+from security.router import router as security_router
+
+# to get a string like this run:
+# openssl rand -hex 32
+
+
+class State(TypedDict):
+    authenticator: BaseAuthenticator
+
+
+@asynccontextmanager
+async def lifespan(_app: FastAPI) -> AsyncIterator[State]:
+    yield {"authenticator": UnsafeAuthenticator()}
+
+
+app = FastAPI(lifespan=lifespan)
+app.include_router(security_router)
+
+
+@app.get("/users/me/")
+async def read_users_me(
+    current_user: Annotated[UserInfo, Depends(get_user)],
+) -> UserInfo:
+    return current_user
+
+
+@app.get("/users/me/premium")
+async def read_own_items(
+    current_user: Annotated[
+        UserInfo, Depends(GetUserWithRole("premium"))
+    ],
+):
+    return current_user

Chapter08/secured-api/main.py

@@ -0,0 +1,23 @@
+[project]
+name = "secured-api"
+version = "0.1.0"
+description = "Add your description here"
+readme = "README.md"
+requires-python = ">=3.13"
+dependencies = [
+    "fastapi[standard]>=0.135.3",
+    "pwdlib[argon2]>=0.3.0",
+    "pyjwt>=2.12.1",
+]
+
+[dependency-groups]
+dev = [
+    "ruff>=0.15.10",
+    "ty>=0.0.29",
+]
+
+[tool.ruff]
+line-length = 61
+
+[tool.ruff.lint]
+select = ["I", "E", "F", "Q", "UP", "FAST", "ARG"]

Chapter08/secured-api/pyproject.toml

@@ -0,0 +1,4 @@
+from .base import BaseAuthenticator
+from .unsafe import UnsafeAuthenticator
+
+__all__ = ["BaseAuthenticator", "UnsafeAuthenticator"]

Chapter08/secured-api/security/__init__.py

@@ -0,0 +1,23 @@
+from abc import ABC, abstractmethod
+
+from ..commons import UserInfo
+
+
+class BaseAuthenticator(ABC):
+    @abstractmethod
+    async def verify_user_and_password(
+        self, username: str, password: str
+    ) -> UserInfo | None:
+        pass
+
+    @abstractmethod
+    async def generate_user_token(
+        self, username: str, password: str
+    ) -> str:
+        pass
+
+    @abstractmethod
+    async def resolve_token(
+        self, token: str
+    ) -> UserInfo | None:
+        pass

Chapter08/secured-api/security/authenticators/__init__.py

@@ -0,0 +1,26 @@
+from ..commons import UserInfo
+from .base import BaseAuthenticator
+
+
+class UnsafeAuthenticator(BaseAuthenticator):
+    async def verify_user_and_password(
+        self, username: str, password: str
+    ) -> UserInfo:
+        _ = password
+        return UserInfo(
+            username=username, roles={"standard", "premium"}
+        )
+
+    async def generate_user_token(
+        self, username, password
+    ) -> str:
+        user = await self.verify_user_and_password(
+            username, password
+        )
+        return f"tokenized{user.username}"
+
+    async def resolve_token(self, token) -> UserInfo | None:
+        if token.startswith("tokenized"):
+            return UserInfo(
+                username=token.removeprefix("tokenized")
+            )

Chapter08/secured-api/security/authenticators/base.py

@@ -0,0 +1,10 @@
+from typing import Literal
+
+from pydantic import BaseModel
+
+Role = Literal["standard", "premium", "gold"]
+
+
+class UserInfo(BaseModel):
+    username: str
+    roles: set[Role] = {"standard"}

Chapter08/secured-api/security/authenticators/unsafe.py

@@ -0,0 +1,51 @@
+from typing import Annotated
+
+from fastapi import Depends, HTTPException, Request, status
+from fastapi.security import OAuth2PasswordBearer
+
+from .authenticator import BaseAuthenticator
+from .commons import UserInfo
+
+oauth2_scheme = OAuth2PasswordBearer(
+    tokenUrl="token",
+    scopes={
+        "me": "Read information about the current user.",
+        "items": "Read items.",
+    },
+)
+
+
+async def get_authenticator(
+    request: Request,
+) -> BaseAuthenticator:
+    return request.state.authenticator
+
+
+async def get_user(
+    token: Annotated[str, Depends(oauth2_scheme)],
+    authenticator: Annotated[
+        BaseAuthenticator, Depends(get_authenticator)
+    ],
+) -> UserInfo:
+    user = await authenticator.resolve_token(token)
+    if not user:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Invalid token",
+        )
+    return user
+
+
+class GetUserWithRole:
+    def __init__(self, role: str):
+        self.role = role
+
+    async def __call__(
+        self, user: Annotated[UserInfo, Depends(get_user)]
+    ):
+        if self.role not in user.roles:
+            raise HTTPException(
+                status_code=status.HTTP_403_FORBIDDEN,
+                detail="Insufficient permission",
+            )
+        return user