Sync fastapi docs from 11614be9 on 2026-03-11
This commit is contained in:
The Librarian
@@ -1,32 +0,0 @@
|
||||
from typing import Annotated, Union
|
||||
|
||||
from fastapi import Depends, FastAPI
|
||||
from fastapi.security import OAuth2PasswordBearer
|
||||
from pydantic import BaseModel
|
||||
|
||||
app = FastAPI()
|
||||
|
||||
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
|
||||
|
||||
|
||||
class User(BaseModel):
|
||||
username: str
|
||||
email: Union[str, None] = None
|
||||
full_name: Union[str, None] = None
|
||||
disabled: Union[bool, None] = None
|
||||
|
||||
|
||||
def fake_decode_token(token):
|
||||
return User(
|
||||
username=token + "fakedecoded", email="john@example.com", full_name="John Doe"
|
||||
)
|
||||
|
||||
|
||||
async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]):
|
||||
user = fake_decode_token(token)
|
||||
return user
|
||||
|
||||
|
||||
@app.get("/users/me")
|
||||
async def read_users_me(current_user: Annotated[User, Depends(get_current_user)]):
|
||||
return current_user
|
||||
@@ -1,32 +0,0 @@
|
||||
from typing import Union
|
||||
|
||||
from fastapi import Depends, FastAPI
|
||||
from fastapi.security import OAuth2PasswordBearer
|
||||
from pydantic import BaseModel
|
||||
|
||||
app = FastAPI()
|
||||
|
||||
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
|
||||
|
||||
|
||||
class User(BaseModel):
|
||||
username: str
|
||||
email: Union[str, None] = None
|
||||
full_name: Union[str, None] = None
|
||||
disabled: Union[bool, None] = None
|
||||
|
||||
|
||||
def fake_decode_token(token):
|
||||
return User(
|
||||
username=token + "fakedecoded", email="john@example.com", full_name="John Doe"
|
||||
)
|
||||
|
||||
|
||||
async def get_current_user(token: str = Depends(oauth2_scheme)):
|
||||
user = fake_decode_token(token)
|
||||
return user
|
||||
|
||||
|
||||
@app.get("/users/me")
|
||||
async def read_users_me(current_user: User = Depends(get_current_user)):
|
||||
return current_user
|
||||
@@ -1,94 +0,0 @@
|
||||
from typing import Annotated, Union
|
||||
|
||||
from fastapi import Depends, FastAPI, HTTPException, status
|
||||
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
|
||||
from pydantic import BaseModel
|
||||
|
||||
fake_users_db = {
|
||||
"johndoe": {
|
||||
"username": "johndoe",
|
||||
"full_name": "John Doe",
|
||||
"email": "johndoe@example.com",
|
||||
"hashed_password": "fakehashedsecret",
|
||||
"disabled": False,
|
||||
},
|
||||
"alice": {
|
||||
"username": "alice",
|
||||
"full_name": "Alice Wonderson",
|
||||
"email": "alice@example.com",
|
||||
"hashed_password": "fakehashedsecret2",
|
||||
"disabled": True,
|
||||
},
|
||||
}
|
||||
|
||||
app = FastAPI()
|
||||
|
||||
|
||||
def fake_hash_password(password: str):
|
||||
return "fakehashed" + password
|
||||
|
||||
|
||||
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
|
||||
|
||||
|
||||
class User(BaseModel):
|
||||
username: str
|
||||
email: Union[str, None] = None
|
||||
full_name: Union[str, None] = None
|
||||
disabled: Union[bool, None] = None
|
||||
|
||||
|
||||
class UserInDB(User):
|
||||
hashed_password: str
|
||||
|
||||
|
||||
def get_user(db, username: str):
|
||||
if username in db:
|
||||
user_dict = db[username]
|
||||
return UserInDB(**user_dict)
|
||||
|
||||
|
||||
def fake_decode_token(token):
|
||||
# This doesn't provide any security at all
|
||||
# Check the next version
|
||||
user = get_user(fake_users_db, token)
|
||||
return user
|
||||
|
||||
|
||||
async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]):
|
||||
user = fake_decode_token(token)
|
||||
if not user:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="Not authenticated",
|
||||
headers={"WWW-Authenticate": "Bearer"},
|
||||
)
|
||||
return user
|
||||
|
||||
|
||||
async def get_current_active_user(
|
||||
current_user: Annotated[User, Depends(get_current_user)],
|
||||
):
|
||||
if current_user.disabled:
|
||||
raise HTTPException(status_code=400, detail="Inactive user")
|
||||
return current_user
|
||||
|
||||
|
||||
@app.post("/token")
|
||||
async def login(form_data: Annotated[OAuth2PasswordRequestForm, Depends()]):
|
||||
user_dict = fake_users_db.get(form_data.username)
|
||||
if not user_dict:
|
||||
raise HTTPException(status_code=400, detail="Incorrect username or password")
|
||||
user = UserInDB(**user_dict)
|
||||
hashed_password = fake_hash_password(form_data.password)
|
||||
if not hashed_password == user.hashed_password:
|
||||
raise HTTPException(status_code=400, detail="Incorrect username or password")
|
||||
|
||||
return {"access_token": user.username, "token_type": "bearer"}
|
||||
|
||||
|
||||
@app.get("/users/me")
|
||||
async def read_users_me(
|
||||
current_user: Annotated[User, Depends(get_current_active_user)],
|
||||
):
|
||||
return current_user
|
||||
@@ -1,90 +0,0 @@
|
||||
from typing import Union
|
||||
|
||||
from fastapi import Depends, FastAPI, HTTPException, status
|
||||
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
|
||||
from pydantic import BaseModel
|
||||
|
||||
fake_users_db = {
|
||||
"johndoe": {
|
||||
"username": "johndoe",
|
||||
"full_name": "John Doe",
|
||||
"email": "johndoe@example.com",
|
||||
"hashed_password": "fakehashedsecret",
|
||||
"disabled": False,
|
||||
},
|
||||
"alice": {
|
||||
"username": "alice",
|
||||
"full_name": "Alice Wonderson",
|
||||
"email": "alice@example.com",
|
||||
"hashed_password": "fakehashedsecret2",
|
||||
"disabled": True,
|
||||
},
|
||||
}
|
||||
|
||||
app = FastAPI()
|
||||
|
||||
|
||||
def fake_hash_password(password: str):
|
||||
return "fakehashed" + password
|
||||
|
||||
|
||||
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
|
||||
|
||||
|
||||
class User(BaseModel):
|
||||
username: str
|
||||
email: Union[str, None] = None
|
||||
full_name: Union[str, None] = None
|
||||
disabled: Union[bool, None] = None
|
||||
|
||||
|
||||
class UserInDB(User):
|
||||
hashed_password: str
|
||||
|
||||
|
||||
def get_user(db, username: str):
|
||||
if username in db:
|
||||
user_dict = db[username]
|
||||
return UserInDB(**user_dict)
|
||||
|
||||
|
||||
def fake_decode_token(token):
|
||||
# This doesn't provide any security at all
|
||||
# Check the next version
|
||||
user = get_user(fake_users_db, token)
|
||||
return user
|
||||
|
||||
|
||||
async def get_current_user(token: str = Depends(oauth2_scheme)):
|
||||
user = fake_decode_token(token)
|
||||
if not user:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="Not authenticated",
|
||||
headers={"WWW-Authenticate": "Bearer"},
|
||||
)
|
||||
return user
|
||||
|
||||
|
||||
async def get_current_active_user(current_user: User = Depends(get_current_user)):
|
||||
if current_user.disabled:
|
||||
raise HTTPException(status_code=400, detail="Inactive user")
|
||||
return current_user
|
||||
|
||||
|
||||
@app.post("/token")
|
||||
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
|
||||
user_dict = fake_users_db.get(form_data.username)
|
||||
if not user_dict:
|
||||
raise HTTPException(status_code=400, detail="Incorrect username or password")
|
||||
user = UserInDB(**user_dict)
|
||||
hashed_password = fake_hash_password(form_data.password)
|
||||
if not hashed_password == user.hashed_password:
|
||||
raise HTTPException(status_code=400, detail="Incorrect username or password")
|
||||
|
||||
return {"access_token": user.username, "token_type": "bearer"}
|
||||
|
||||
|
||||
@app.get("/users/me")
|
||||
async def read_users_me(current_user: User = Depends(get_current_active_user)):
|
||||
return current_user
|
||||
@@ -48,6 +48,8 @@ class UserInDB(User):
|
||||
|
||||
password_hash = PasswordHash.recommended()
|
||||
|
||||
DUMMY_HASH = password_hash.hash("dummypassword")
|
||||
|
||||
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
|
||||
|
||||
app = FastAPI()
|
||||
@@ -70,6 +72,7 @@ def get_user(db, username: str):
|
||||
def authenticate_user(fake_db, username: str, password: str):
|
||||
user = get_user(fake_db, username)
|
||||
if not user:
|
||||
verify_password(password, DUMMY_HASH)
|
||||
return False
|
||||
if not verify_password(password, user.hashed_password):
|
||||
return False
|
||||
@@ -133,10 +136,10 @@ async def login_for_access_token(
|
||||
return Token(access_token=access_token, token_type="bearer")
|
||||
|
||||
|
||||
@app.get("/users/me/", response_model=User)
|
||||
@app.get("/users/me/")
|
||||
async def read_users_me(
|
||||
current_user: Annotated[User, Depends(get_current_active_user)],
|
||||
):
|
||||
) -> User:
|
||||
return current_user
|
||||
|
||||
|
||||
|
||||
@@ -1,147 +0,0 @@
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from typing import Annotated, Union
|
||||
|
||||
import jwt
|
||||
from fastapi import Depends, FastAPI, HTTPException, status
|
||||
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
|
||||
from jwt.exceptions import InvalidTokenError
|
||||
from pwdlib import PasswordHash
|
||||
from pydantic import BaseModel
|
||||
|
||||
# to get a string like this run:
|
||||
# openssl rand -hex 32
|
||||
SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
|
||||
ALGORITHM = "HS256"
|
||||
ACCESS_TOKEN_EXPIRE_MINUTES = 30
|
||||
|
||||
|
||||
fake_users_db = {
|
||||
"johndoe": {
|
||||
"username": "johndoe",
|
||||
"full_name": "John Doe",
|
||||
"email": "johndoe@example.com",
|
||||
"hashed_password": "$argon2id$v=19$m=65536,t=3,p=4$wagCPXjifgvUFBzq4hqe3w$CYaIb8sB+wtD+Vu/P4uod1+Qof8h+1g7bbDlBID48Rc",
|
||||
"disabled": False,
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
class Token(BaseModel):
|
||||
access_token: str
|
||||
token_type: str
|
||||
|
||||
|
||||
class TokenData(BaseModel):
|
||||
username: Union[str, None] = None
|
||||
|
||||
|
||||
class User(BaseModel):
|
||||
username: str
|
||||
email: Union[str, None] = None
|
||||
full_name: Union[str, None] = None
|
||||
disabled: Union[bool, None] = None
|
||||
|
||||
|
||||
class UserInDB(User):
|
||||
hashed_password: str
|
||||
|
||||
|
||||
password_hash = PasswordHash.recommended()
|
||||
|
||||
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
|
||||
|
||||
app = FastAPI()
|
||||
|
||||
|
||||
def verify_password(plain_password, hashed_password):
|
||||
return password_hash.verify(plain_password, hashed_password)
|
||||
|
||||
|
||||
def get_password_hash(password):
|
||||
return password_hash.hash(password)
|
||||
|
||||
|
||||
def get_user(db, username: str):
|
||||
if username in db:
|
||||
user_dict = db[username]
|
||||
return UserInDB(**user_dict)
|
||||
|
||||
|
||||
def authenticate_user(fake_db, username: str, password: str):
|
||||
user = get_user(fake_db, username)
|
||||
if not user:
|
||||
return False
|
||||
if not verify_password(password, user.hashed_password):
|
||||
return False
|
||||
return user
|
||||
|
||||
|
||||
def create_access_token(data: dict, expires_delta: Union[timedelta, None] = None):
|
||||
to_encode = data.copy()
|
||||
if expires_delta:
|
||||
expire = datetime.now(timezone.utc) + expires_delta
|
||||
else:
|
||||
expire = datetime.now(timezone.utc) + timedelta(minutes=15)
|
||||
to_encode.update({"exp": expire})
|
||||
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
|
||||
return encoded_jwt
|
||||
|
||||
|
||||
async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]):
|
||||
credentials_exception = HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="Could not validate credentials",
|
||||
headers={"WWW-Authenticate": "Bearer"},
|
||||
)
|
||||
try:
|
||||
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
|
||||
username = payload.get("sub")
|
||||
if username is None:
|
||||
raise credentials_exception
|
||||
token_data = TokenData(username=username)
|
||||
except InvalidTokenError:
|
||||
raise credentials_exception
|
||||
user = get_user(fake_users_db, username=token_data.username)
|
||||
if user is None:
|
||||
raise credentials_exception
|
||||
return user
|
||||
|
||||
|
||||
async def get_current_active_user(
|
||||
current_user: Annotated[User, Depends(get_current_user)],
|
||||
):
|
||||
if current_user.disabled:
|
||||
raise HTTPException(status_code=400, detail="Inactive user")
|
||||
return current_user
|
||||
|
||||
|
||||
@app.post("/token")
|
||||
async def login_for_access_token(
|
||||
form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
|
||||
) -> Token:
|
||||
user = authenticate_user(fake_users_db, form_data.username, form_data.password)
|
||||
if not user:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="Incorrect username or password",
|
||||
headers={"WWW-Authenticate": "Bearer"},
|
||||
)
|
||||
access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
|
||||
access_token = create_access_token(
|
||||
data={"sub": user.username}, expires_delta=access_token_expires
|
||||
)
|
||||
return Token(access_token=access_token, token_type="bearer")
|
||||
|
||||
|
||||
@app.get("/users/me/", response_model=User)
|
||||
async def read_users_me(
|
||||
current_user: Annotated[User, Depends(get_current_active_user)],
|
||||
):
|
||||
return current_user
|
||||
|
||||
|
||||
@app.get("/users/me/items/")
|
||||
async def read_own_items(
|
||||
current_user: Annotated[User, Depends(get_current_active_user)],
|
||||
):
|
||||
return [{"item_id": "Foo", "owner": current_user.username}]
|
||||
@@ -47,6 +47,8 @@ class UserInDB(User):
|
||||
|
||||
password_hash = PasswordHash.recommended()
|
||||
|
||||
DUMMY_HASH = password_hash.hash("dummypassword")
|
||||
|
||||
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
|
||||
|
||||
app = FastAPI()
|
||||
@@ -69,6 +71,7 @@ def get_user(db, username: str):
|
||||
def authenticate_user(fake_db, username: str, password: str):
|
||||
user = get_user(fake_db, username)
|
||||
if not user:
|
||||
verify_password(password, DUMMY_HASH)
|
||||
return False
|
||||
if not verify_password(password, user.hashed_password):
|
||||
return False
|
||||
@@ -130,8 +133,8 @@ async def login_for_access_token(
|
||||
return Token(access_token=access_token, token_type="bearer")
|
||||
|
||||
|
||||
@app.get("/users/me/", response_model=User)
|
||||
async def read_users_me(current_user: User = Depends(get_current_active_user)):
|
||||
@app.get("/users/me/")
|
||||
async def read_users_me(current_user: User = Depends(get_current_active_user)) -> User:
|
||||
return current_user
|
||||
|
||||
|
||||
|
||||
@@ -1,141 +0,0 @@
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from typing import Union
|
||||
|
||||
import jwt
|
||||
from fastapi import Depends, FastAPI, HTTPException, status
|
||||
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
|
||||
from jwt.exceptions import InvalidTokenError
|
||||
from pwdlib import PasswordHash
|
||||
from pydantic import BaseModel
|
||||
|
||||
# to get a string like this run:
|
||||
# openssl rand -hex 32
|
||||
SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
|
||||
ALGORITHM = "HS256"
|
||||
ACCESS_TOKEN_EXPIRE_MINUTES = 30
|
||||
|
||||
|
||||
fake_users_db = {
|
||||
"johndoe": {
|
||||
"username": "johndoe",
|
||||
"full_name": "John Doe",
|
||||
"email": "johndoe@example.com",
|
||||
"hashed_password": "$argon2id$v=19$m=65536,t=3,p=4$wagCPXjifgvUFBzq4hqe3w$CYaIb8sB+wtD+Vu/P4uod1+Qof8h+1g7bbDlBID48Rc",
|
||||
"disabled": False,
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
class Token(BaseModel):
|
||||
access_token: str
|
||||
token_type: str
|
||||
|
||||
|
||||
class TokenData(BaseModel):
|
||||
username: Union[str, None] = None
|
||||
|
||||
|
||||
class User(BaseModel):
|
||||
username: str
|
||||
email: Union[str, None] = None
|
||||
full_name: Union[str, None] = None
|
||||
disabled: Union[bool, None] = None
|
||||
|
||||
|
||||
class UserInDB(User):
|
||||
hashed_password: str
|
||||
|
||||
|
||||
password_hash = PasswordHash.recommended()
|
||||
|
||||
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
|
||||
|
||||
app = FastAPI()
|
||||
|
||||
|
||||
def verify_password(plain_password, hashed_password):
|
||||
return password_hash.verify(plain_password, hashed_password)
|
||||
|
||||
|
||||
def get_password_hash(password):
|
||||
return password_hash.hash(password)
|
||||
|
||||
|
||||
def get_user(db, username: str):
|
||||
if username in db:
|
||||
user_dict = db[username]
|
||||
return UserInDB(**user_dict)
|
||||
|
||||
|
||||
def authenticate_user(fake_db, username: str, password: str):
|
||||
user = get_user(fake_db, username)
|
||||
if not user:
|
||||
return False
|
||||
if not verify_password(password, user.hashed_password):
|
||||
return False
|
||||
return user
|
||||
|
||||
|
||||
def create_access_token(data: dict, expires_delta: Union[timedelta, None] = None):
|
||||
to_encode = data.copy()
|
||||
if expires_delta:
|
||||
expire = datetime.now(timezone.utc) + expires_delta
|
||||
else:
|
||||
expire = datetime.now(timezone.utc) + timedelta(minutes=15)
|
||||
to_encode.update({"exp": expire})
|
||||
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
|
||||
return encoded_jwt
|
||||
|
||||
|
||||
async def get_current_user(token: str = Depends(oauth2_scheme)):
|
||||
credentials_exception = HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="Could not validate credentials",
|
||||
headers={"WWW-Authenticate": "Bearer"},
|
||||
)
|
||||
try:
|
||||
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
|
||||
username = payload.get("sub")
|
||||
if username is None:
|
||||
raise credentials_exception
|
||||
token_data = TokenData(username=username)
|
||||
except InvalidTokenError:
|
||||
raise credentials_exception
|
||||
user = get_user(fake_users_db, username=token_data.username)
|
||||
if user is None:
|
||||
raise credentials_exception
|
||||
return user
|
||||
|
||||
|
||||
async def get_current_active_user(current_user: User = Depends(get_current_user)):
|
||||
if current_user.disabled:
|
||||
raise HTTPException(status_code=400, detail="Inactive user")
|
||||
return current_user
|
||||
|
||||
|
||||
@app.post("/token")
|
||||
async def login_for_access_token(
|
||||
form_data: OAuth2PasswordRequestForm = Depends(),
|
||||
) -> Token:
|
||||
user = authenticate_user(fake_users_db, form_data.username, form_data.password)
|
||||
if not user:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="Incorrect username or password",
|
||||
headers={"WWW-Authenticate": "Bearer"},
|
||||
)
|
||||
access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
|
||||
access_token = create_access_token(
|
||||
data={"sub": user.username}, expires_delta=access_token_expires
|
||||
)
|
||||
return Token(access_token=access_token, token_type="bearer")
|
||||
|
||||
|
||||
@app.get("/users/me/", response_model=User)
|
||||
async def read_users_me(current_user: User = Depends(get_current_active_user)):
|
||||
return current_user
|
||||
|
||||
|
||||
@app.get("/users/me/items/")
|
||||
async def read_own_items(current_user: User = Depends(get_current_active_user)):
|
||||
return [{"item_id": "Foo", "owner": current_user.username}]
|
||||
@@ -60,6 +60,8 @@ class UserInDB(User):
|
||||
|
||||
password_hash = PasswordHash.recommended()
|
||||
|
||||
DUMMY_HASH = password_hash.hash("dummypassword")
|
||||
|
||||
oauth2_scheme = OAuth2PasswordBearer(
|
||||
tokenUrl="token",
|
||||
scopes={"me": "Read information about the current user.", "items": "Read items."},
|
||||
@@ -85,6 +87,7 @@ def get_user(db, username: str):
|
||||
def authenticate_user(fake_db, username: str, password: str):
|
||||
user = get_user(fake_db, username)
|
||||
if not user:
|
||||
verify_password(password, DUMMY_HASH)
|
||||
return False
|
||||
if not verify_password(password, user.hashed_password):
|
||||
return False
|
||||
@@ -160,10 +163,10 @@ async def login_for_access_token(
|
||||
return Token(access_token=access_token, token_type="bearer")
|
||||
|
||||
|
||||
@app.get("/users/me/", response_model=User)
|
||||
@app.get("/users/me/")
|
||||
async def read_users_me(
|
||||
current_user: Annotated[User, Depends(get_current_active_user)],
|
||||
):
|
||||
) -> User:
|
||||
return current_user
|
||||
|
||||
|
||||
|
||||
@@ -1,179 +0,0 @@
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from typing import Annotated, Union
|
||||
|
||||
import jwt
|
||||
from fastapi import Depends, FastAPI, HTTPException, Security, status
|
||||
from fastapi.security import (
|
||||
OAuth2PasswordBearer,
|
||||
OAuth2PasswordRequestForm,
|
||||
SecurityScopes,
|
||||
)
|
||||
from jwt.exceptions import InvalidTokenError
|
||||
from pwdlib import PasswordHash
|
||||
from pydantic import BaseModel, ValidationError
|
||||
|
||||
# to get a string like this run:
|
||||
# openssl rand -hex 32
|
||||
SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
|
||||
ALGORITHM = "HS256"
|
||||
ACCESS_TOKEN_EXPIRE_MINUTES = 30
|
||||
|
||||
|
||||
fake_users_db = {
|
||||
"johndoe": {
|
||||
"username": "johndoe",
|
||||
"full_name": "John Doe",
|
||||
"email": "johndoe@example.com",
|
||||
"hashed_password": "$argon2id$v=19$m=65536,t=3,p=4$wagCPXjifgvUFBzq4hqe3w$CYaIb8sB+wtD+Vu/P4uod1+Qof8h+1g7bbDlBID48Rc",
|
||||
"disabled": False,
|
||||
},
|
||||
"alice": {
|
||||
"username": "alice",
|
||||
"full_name": "Alice Chains",
|
||||
"email": "alicechains@example.com",
|
||||
"hashed_password": "$argon2id$v=19$m=65536,t=3,p=4$g2/AV1zwopqUntPKJavBFw$BwpRGDCyUHLvHICnwijyX8ROGoiUPwNKZ7915MeYfCE",
|
||||
"disabled": True,
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
class Token(BaseModel):
|
||||
access_token: str
|
||||
token_type: str
|
||||
|
||||
|
||||
class TokenData(BaseModel):
|
||||
username: Union[str, None] = None
|
||||
scopes: list[str] = []
|
||||
|
||||
|
||||
class User(BaseModel):
|
||||
username: str
|
||||
email: Union[str, None] = None
|
||||
full_name: Union[str, None] = None
|
||||
disabled: Union[bool, None] = None
|
||||
|
||||
|
||||
class UserInDB(User):
|
||||
hashed_password: str
|
||||
|
||||
|
||||
password_hash = PasswordHash.recommended()
|
||||
|
||||
oauth2_scheme = OAuth2PasswordBearer(
|
||||
tokenUrl="token",
|
||||
scopes={"me": "Read information about the current user.", "items": "Read items."},
|
||||
)
|
||||
|
||||
app = FastAPI()
|
||||
|
||||
|
||||
def verify_password(plain_password, hashed_password):
|
||||
return password_hash.verify(plain_password, hashed_password)
|
||||
|
||||
|
||||
def get_password_hash(password):
|
||||
return password_hash.hash(password)
|
||||
|
||||
|
||||
def get_user(db, username: str):
|
||||
if username in db:
|
||||
user_dict = db[username]
|
||||
return UserInDB(**user_dict)
|
||||
|
||||
|
||||
def authenticate_user(fake_db, username: str, password: str):
|
||||
user = get_user(fake_db, username)
|
||||
if not user:
|
||||
return False
|
||||
if not verify_password(password, user.hashed_password):
|
||||
return False
|
||||
return user
|
||||
|
||||
|
||||
def create_access_token(data: dict, expires_delta: Union[timedelta, None] = None):
|
||||
to_encode = data.copy()
|
||||
if expires_delta:
|
||||
expire = datetime.now(timezone.utc) + expires_delta
|
||||
else:
|
||||
expire = datetime.now(timezone.utc) + timedelta(minutes=15)
|
||||
to_encode.update({"exp": expire})
|
||||
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
|
||||
return encoded_jwt
|
||||
|
||||
|
||||
async def get_current_user(
|
||||
security_scopes: SecurityScopes, token: Annotated[str, Depends(oauth2_scheme)]
|
||||
):
|
||||
if security_scopes.scopes:
|
||||
authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
|
||||
else:
|
||||
authenticate_value = "Bearer"
|
||||
credentials_exception = HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="Could not validate credentials",
|
||||
headers={"WWW-Authenticate": authenticate_value},
|
||||
)
|
||||
try:
|
||||
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
|
||||
username = payload.get("sub")
|
||||
if username is None:
|
||||
raise credentials_exception
|
||||
scope: str = payload.get("scope", "")
|
||||
token_scopes = scope.split(" ")
|
||||
token_data = TokenData(scopes=token_scopes, username=username)
|
||||
except (InvalidTokenError, ValidationError):
|
||||
raise credentials_exception
|
||||
user = get_user(fake_users_db, username=token_data.username)
|
||||
if user is None:
|
||||
raise credentials_exception
|
||||
for scope in security_scopes.scopes:
|
||||
if scope not in token_data.scopes:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="Not enough permissions",
|
||||
headers={"WWW-Authenticate": authenticate_value},
|
||||
)
|
||||
return user
|
||||
|
||||
|
||||
async def get_current_active_user(
|
||||
current_user: Annotated[User, Security(get_current_user, scopes=["me"])],
|
||||
):
|
||||
if current_user.disabled:
|
||||
raise HTTPException(status_code=400, detail="Inactive user")
|
||||
return current_user
|
||||
|
||||
|
||||
@app.post("/token")
|
||||
async def login_for_access_token(
|
||||
form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
|
||||
) -> Token:
|
||||
user = authenticate_user(fake_users_db, form_data.username, form_data.password)
|
||||
if not user:
|
||||
raise HTTPException(status_code=400, detail="Incorrect username or password")
|
||||
access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
|
||||
access_token = create_access_token(
|
||||
data={"sub": user.username, "scope": " ".join(form_data.scopes)},
|
||||
expires_delta=access_token_expires,
|
||||
)
|
||||
return Token(access_token=access_token, token_type="bearer")
|
||||
|
||||
|
||||
@app.get("/users/me/", response_model=User)
|
||||
async def read_users_me(
|
||||
current_user: Annotated[User, Depends(get_current_active_user)],
|
||||
):
|
||||
return current_user
|
||||
|
||||
|
||||
@app.get("/users/me/items/")
|
||||
async def read_own_items(
|
||||
current_user: Annotated[User, Security(get_current_active_user, scopes=["items"])],
|
||||
):
|
||||
return [{"item_id": "Foo", "owner": current_user.username}]
|
||||
|
||||
|
||||
@app.get("/status/")
|
||||
async def read_system_status(current_user: Annotated[User, Depends(get_current_user)]):
|
||||
return {"status": "ok"}
|
||||
@@ -59,6 +59,8 @@ class UserInDB(User):
|
||||
|
||||
password_hash = PasswordHash.recommended()
|
||||
|
||||
DUMMY_HASH = password_hash.hash("dummypassword")
|
||||
|
||||
oauth2_scheme = OAuth2PasswordBearer(
|
||||
tokenUrl="token",
|
||||
scopes={"me": "Read information about the current user.", "items": "Read items."},
|
||||
@@ -84,6 +86,7 @@ def get_user(db, username: str):
|
||||
def authenticate_user(fake_db, username: str, password: str):
|
||||
user = get_user(fake_db, username)
|
||||
if not user:
|
||||
verify_password(password, DUMMY_HASH)
|
||||
return False
|
||||
if not verify_password(password, user.hashed_password):
|
||||
return False
|
||||
@@ -159,8 +162,8 @@ async def login_for_access_token(
|
||||
return Token(access_token=access_token, token_type="bearer")
|
||||
|
||||
|
||||
@app.get("/users/me/", response_model=User)
|
||||
async def read_users_me(current_user: User = Depends(get_current_active_user)):
|
||||
@app.get("/users/me/")
|
||||
async def read_users_me(current_user: User = Depends(get_current_active_user)) -> User:
|
||||
return current_user
|
||||
|
||||
|
||||
|
||||
@@ -1,177 +0,0 @@
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from typing import Union
|
||||
|
||||
import jwt
|
||||
from fastapi import Depends, FastAPI, HTTPException, Security, status
|
||||
from fastapi.security import (
|
||||
OAuth2PasswordBearer,
|
||||
OAuth2PasswordRequestForm,
|
||||
SecurityScopes,
|
||||
)
|
||||
from jwt.exceptions import InvalidTokenError
|
||||
from pwdlib import PasswordHash
|
||||
from pydantic import BaseModel, ValidationError
|
||||
|
||||
# to get a string like this run:
|
||||
# openssl rand -hex 32
|
||||
SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
|
||||
ALGORITHM = "HS256"
|
||||
ACCESS_TOKEN_EXPIRE_MINUTES = 30
|
||||
|
||||
|
||||
fake_users_db = {
|
||||
"johndoe": {
|
||||
"username": "johndoe",
|
||||
"full_name": "John Doe",
|
||||
"email": "johndoe@example.com",
|
||||
"hashed_password": "$argon2id$v=19$m=65536,t=3,p=4$wagCPXjifgvUFBzq4hqe3w$CYaIb8sB+wtD+Vu/P4uod1+Qof8h+1g7bbDlBID48Rc",
|
||||
"disabled": False,
|
||||
},
|
||||
"alice": {
|
||||
"username": "alice",
|
||||
"full_name": "Alice Chains",
|
||||
"email": "alicechains@example.com",
|
||||
"hashed_password": "$argon2id$v=19$m=65536,t=3,p=4$g2/AV1zwopqUntPKJavBFw$BwpRGDCyUHLvHICnwijyX8ROGoiUPwNKZ7915MeYfCE",
|
||||
"disabled": True,
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
class Token(BaseModel):
|
||||
access_token: str
|
||||
token_type: str
|
||||
|
||||
|
||||
class TokenData(BaseModel):
|
||||
username: Union[str, None] = None
|
||||
scopes: list[str] = []
|
||||
|
||||
|
||||
class User(BaseModel):
|
||||
username: str
|
||||
email: Union[str, None] = None
|
||||
full_name: Union[str, None] = None
|
||||
disabled: Union[bool, None] = None
|
||||
|
||||
|
||||
class UserInDB(User):
|
||||
hashed_password: str
|
||||
|
||||
|
||||
password_hash = PasswordHash.recommended()
|
||||
|
||||
oauth2_scheme = OAuth2PasswordBearer(
|
||||
tokenUrl="token",
|
||||
scopes={"me": "Read information about the current user.", "items": "Read items."},
|
||||
)
|
||||
|
||||
app = FastAPI()
|
||||
|
||||
|
||||
def verify_password(plain_password, hashed_password):
|
||||
return password_hash.verify(plain_password, hashed_password)
|
||||
|
||||
|
||||
def get_password_hash(password):
|
||||
return password_hash.hash(password)
|
||||
|
||||
|
||||
def get_user(db, username: str):
|
||||
if username in db:
|
||||
user_dict = db[username]
|
||||
return UserInDB(**user_dict)
|
||||
|
||||
|
||||
def authenticate_user(fake_db, username: str, password: str):
|
||||
user = get_user(fake_db, username)
|
||||
if not user:
|
||||
return False
|
||||
if not verify_password(password, user.hashed_password):
|
||||
return False
|
||||
return user
|
||||
|
||||
|
||||
def create_access_token(data: dict, expires_delta: Union[timedelta, None] = None):
|
||||
to_encode = data.copy()
|
||||
if expires_delta:
|
||||
expire = datetime.now(timezone.utc) + expires_delta
|
||||
else:
|
||||
expire = datetime.now(timezone.utc) + timedelta(minutes=15)
|
||||
to_encode.update({"exp": expire})
|
||||
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
|
||||
return encoded_jwt
|
||||
|
||||
|
||||
async def get_current_user(
|
||||
security_scopes: SecurityScopes, token: str = Depends(oauth2_scheme)
|
||||
):
|
||||
if security_scopes.scopes:
|
||||
authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
|
||||
else:
|
||||
authenticate_value = "Bearer"
|
||||
credentials_exception = HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="Could not validate credentials",
|
||||
headers={"WWW-Authenticate": authenticate_value},
|
||||
)
|
||||
try:
|
||||
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
|
||||
username: str = payload.get("sub")
|
||||
if username is None:
|
||||
raise credentials_exception
|
||||
scope: str = payload.get("scope", "")
|
||||
token_scopes = scope.split(" ")
|
||||
token_data = TokenData(scopes=token_scopes, username=username)
|
||||
except (InvalidTokenError, ValidationError):
|
||||
raise credentials_exception
|
||||
user = get_user(fake_users_db, username=token_data.username)
|
||||
if user is None:
|
||||
raise credentials_exception
|
||||
for scope in security_scopes.scopes:
|
||||
if scope not in token_data.scopes:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="Not enough permissions",
|
||||
headers={"WWW-Authenticate": authenticate_value},
|
||||
)
|
||||
return user
|
||||
|
||||
|
||||
async def get_current_active_user(
|
||||
current_user: User = Security(get_current_user, scopes=["me"]),
|
||||
):
|
||||
if current_user.disabled:
|
||||
raise HTTPException(status_code=400, detail="Inactive user")
|
||||
return current_user
|
||||
|
||||
|
||||
@app.post("/token")
|
||||
async def login_for_access_token(
|
||||
form_data: OAuth2PasswordRequestForm = Depends(),
|
||||
) -> Token:
|
||||
user = authenticate_user(fake_users_db, form_data.username, form_data.password)
|
||||
if not user:
|
||||
raise HTTPException(status_code=400, detail="Incorrect username or password")
|
||||
access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
|
||||
access_token = create_access_token(
|
||||
data={"sub": user.username, "scope": " ".join(form_data.scopes)},
|
||||
expires_delta=access_token_expires,
|
||||
)
|
||||
return Token(access_token=access_token, token_type="bearer")
|
||||
|
||||
|
||||
@app.get("/users/me/", response_model=User)
|
||||
async def read_users_me(current_user: User = Depends(get_current_active_user)):
|
||||
return current_user
|
||||
|
||||
|
||||
@app.get("/users/me/items/")
|
||||
async def read_own_items(
|
||||
current_user: User = Security(get_current_active_user, scopes=["items"]),
|
||||
):
|
||||
return [{"item_id": "Foo", "owner": current_user.username}]
|
||||
|
||||
|
||||
@app.get("/status/")
|
||||
async def read_system_status(current_user: User = Depends(get_current_user)):
|
||||
return {"status": "ok"}
|
||||
Reference in New Issue
Block a user