Yurii Motov
7 months ago
18 changed files with 10 additions and 712 deletions
@ -1,12 +0,0 @@ |
|||
from fastapi import Depends, FastAPI |
|||
from fastapi.security import OAuth2PasswordBearer |
|||
from typing_extensions import Annotated |
|||
|
|||
app = FastAPI() |
|||
|
|||
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token") |
|||
|
|||
|
|||
@app.get("/items/") |
|||
async def read_items(token: Annotated[str, Depends(oauth2_scheme)]): |
|||
return {"token": token} |
|||
@ -1,33 +0,0 @@ |
|||
from typing import Union |
|||
|
|||
from fastapi import Depends, FastAPI |
|||
from fastapi.security import OAuth2PasswordBearer |
|||
from pydantic import BaseModel |
|||
from typing_extensions import Annotated |
|||
|
|||
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="[email protected]", 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,95 +0,0 @@ |
|||
from typing import Union |
|||
|
|||
from fastapi import Depends, FastAPI, HTTPException, status |
|||
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm |
|||
from pydantic import BaseModel |
|||
from typing_extensions import Annotated |
|||
|
|||
fake_users_db = { |
|||
"johndoe": { |
|||
"username": "johndoe", |
|||
"full_name": "John Doe", |
|||
"email": "[email protected]", |
|||
"hashed_password": "fakehashedsecret", |
|||
"disabled": False, |
|||
}, |
|||
"alice": { |
|||
"username": "alice", |
|||
"full_name": "Alice Wonderson", |
|||
"email": "[email protected]", |
|||
"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,148 +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 |
|||
from typing_extensions import Annotated |
|||
|
|||
# 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": "[email protected]", |
|||
"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}] |
|||
@ -1,177 +0,0 @@ |
|||
from datetime import datetime, timedelta, timezone |
|||
from typing import List, 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": "[email protected]", |
|||
"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": "[email protected]", |
|||
"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"} |
|||
@ -1,180 +0,0 @@ |
|||
from datetime import datetime, timedelta, timezone |
|||
from typing import List, 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 |
|||
from typing_extensions import Annotated |
|||
|
|||
# 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": "[email protected]", |
|||
"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": "[email protected]", |
|||
"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"} |
|||
@ -1,12 +0,0 @@ |
|||
from fastapi import Depends, FastAPI |
|||
from fastapi.security import HTTPBasic, HTTPBasicCredentials |
|||
from typing_extensions import Annotated |
|||
|
|||
app = FastAPI() |
|||
|
|||
security = HTTPBasic() |
|||
|
|||
|
|||
@app.get("/users/me") |
|||
def read_current_user(credentials: Annotated[HTTPBasicCredentials, Depends(security)]): |
|||
return {"username": credentials.username, "password": credentials.password} |
|||
@ -1,36 +0,0 @@ |
|||
import secrets |
|||
|
|||
from fastapi import Depends, FastAPI, HTTPException, status |
|||
from fastapi.security import HTTPBasic, HTTPBasicCredentials |
|||
from typing_extensions import Annotated |
|||
|
|||
app = FastAPI() |
|||
|
|||
security = HTTPBasic() |
|||
|
|||
|
|||
def get_current_username( |
|||
credentials: Annotated[HTTPBasicCredentials, Depends(security)], |
|||
): |
|||
current_username_bytes = credentials.username.encode("utf8") |
|||
correct_username_bytes = b"stanleyjobson" |
|||
is_correct_username = secrets.compare_digest( |
|||
current_username_bytes, correct_username_bytes |
|||
) |
|||
current_password_bytes = credentials.password.encode("utf8") |
|||
correct_password_bytes = b"swordfish" |
|||
is_correct_password = secrets.compare_digest( |
|||
current_password_bytes, correct_password_bytes |
|||
) |
|||
if not (is_correct_username and is_correct_password): |
|||
raise HTTPException( |
|||
status_code=status.HTTP_401_UNAUTHORIZED, |
|||
detail="Incorrect username or password", |
|||
headers={"WWW-Authenticate": "Basic"}, |
|||
) |
|||
return credentials.username |
|||
|
|||
|
|||
@app.get("/users/me") |
|||
def read_current_user(username: Annotated[str, Depends(get_current_username)]): |
|||
return {"username": username} |
|||
Loading…
Reference in new issue