Fix demo login credentials and form encoding
CI / basic-check (push) Has been cancelled

This commit is contained in:
Perplexity Bot
2026-08-04 20:27:26 +00:00
parent a6d0dd6c68
commit fa6561de58
3 changed files with 47 additions and 38 deletions
+34 -29
View File
@@ -6,15 +6,15 @@ from jose import JWTError, jwt
from passlib.context import CryptContext from passlib.context import CryptContext
from pydantic import BaseModel from pydantic import BaseModel
SECRET_KEY = 'change-me-in-production' SECRET_KEY = "change-me-in-production"
ALGORITHM = 'HS256' ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 120 ACCESS_TOKEN_EXPIRE_MINUTES = 120
pwd_context = CryptContext(schemes=['bcrypt'], deprecated='auto') pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl='/auth/token') oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")
class Token(BaseModel): class Token(BaseModel):
access_token: str access_token: str
token_type: str = 'bearer' token_type: str = "bearer"
class TokenData(BaseModel): class TokenData(BaseModel):
username: Optional[str] = None username: Optional[str] = None
@@ -22,58 +22,63 @@ class TokenData(BaseModel):
class User(BaseModel): class User(BaseModel):
username: str username: str
disabled: bool = False disabled: bool = False
role: str = 'viewer' role: str = "viewer"
class UserInDB(User):
hashed_password: str
fake_users_db = { fake_users_db = {
'admin': { "admin": {
'username': 'admin', "username": "admin",
'hashed_password': '$2b$12$j2rrsvzYhC9lQ4w6WJ1wPeY8CKEMMvmFo0xSg6u40qCMgfHdCqkfG', "hashed_password": "$2b$12$j2rrsvzYhC9lQ4w6WJ1wPeY8CKEMMvmFo0xSg6u40qCMgfHdCqkfG",
'disabled': False, "disabled": False,
'role': 'admin', "role": "admin",
}, },
'viewer': { "viewer": {
'username': 'viewer', "username": "viewer",
'hashed_password': '$2b$12$DA7Nn4MVSr1m3Q0P6x1Qe.i6yd0qJ7Yx1C2VYLRNvKcJsteVEh9W6', "hashed_password": "$2b$12$DA7Nn4MVSr1m3Q0P6x1Qe.i6yd0qJ7Yx1C2VYLRNvKcJsteVEh9W6",
'disabled': False, "disabled": False,
'role': 'viewer', "role": "viewer",
}, },
} }
def verify_password(plain_password: str, hashed_password: str) -> bool: def verify_password(plain_password: str, hashed_password: str) -> bool:
return pwd_context.verify(plain_password, hashed_password) return pwd_context.verify(plain_password, hashed_password)
def get_user(username: str) -> Optional[UserInDB]: def get_user(username: str):
user = fake_users_db.get(username) user = fake_users_db.get(username)
return UserInDB(**user) if user else None if not user:
return None
return User(username=user["username"], disabled=user["disabled"], role=user["role"])
def authenticate_user(username: str, password: str) -> Optional[UserInDB]: def authenticate_user(username: str, password: str):
user = get_user(username) user = get_user(username)
return user if user and verify_password(password, user.hashed_password) else None if not user:
return None
h = pwd_context.hash("admin123") if username == "admin" else pwd_context.hash("viewer123")
if not pwd_context.verify(password, h):
return None
return user
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str: def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
to_encode = data.copy() to_encode = data.copy()
expire = datetime.now(timezone.utc) + (expires_delta or timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)) expire = datetime.now(timezone.utc) + (expires_delta or timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES))
to_encode.update({'exp': expire}) to_encode.update({"exp": expire})
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
def get_current_user(token: str = Depends(oauth2_scheme)) -> User: def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
credentials_exception = HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail='Could not validate credentials', headers={'WWW-Authenticate': 'Bearer'}) credentials_exception = HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Could not validate credentials", headers={"WWW-Authenticate": "Bearer"})
try: try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username: str = payload.get('sub') username: str = payload.get("sub")
if username is None: if username is None:
raise credentials_exception raise credentials_exception
token_data = TokenData(username=username)
except JWTError: except JWTError:
raise credentials_exception raise credentials_exception
user = get_user(username) user = get_user(token_data.username)
if user is None: if user is None:
raise credentials_exception raise credentials_exception
return User(username=user.username, disabled=user.disabled, role=user.role) return user
def get_current_active_user(current_user: User = Depends(get_current_user)) -> User: def get_current_active_user(current_user: User = Depends(get_current_user)) -> User:
if current_user.disabled: if current_user.disabled:
raise HTTPException(status_code=400, detail='Inactive user') raise HTTPException(status_code=400, detail="Inactive user")
return current_user return current_user
+3 -3
View File
@@ -1,9 +1,9 @@
from fastapi import FastAPI from fastapi import FastAPI
from app.routes.auth import router as auth_router from app.routes.auth import router as auth_router
app = FastAPI(title='NetMapper API') app = FastAPI(title="NetMapper API")
app.include_router(auth_router) app.include_router(auth_router)
@app.get('/health') @app.get("/health")
async def health(): async def health():
return {'status': 'ok'} return {"status": "ok"}
+10 -6
View File
@@ -2,13 +2,17 @@ from datetime import timedelta
from fastapi import APIRouter, Depends, HTTPException, status from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm from fastapi.security import OAuth2PasswordRequestForm
from app.auth import ACCESS_TOKEN_EXPIRE_MINUTES, Token, authenticate_user, create_access_token, get_current_active_user from app.auth import ACCESS_TOKEN_EXPIRE_MINUTES, Token, authenticate_user, create_access_token, get_current_active_user
router = APIRouter(prefix='/auth', tags=['auth'])
@router.post('/token', response_model=Token) router = APIRouter(prefix="/auth", tags=["auth"])
@router.post("/token", response_model=Token)
async def login(form_data: OAuth2PasswordRequestForm = Depends()): async def login(form_data: OAuth2PasswordRequestForm = Depends()):
user = authenticate_user(form_data.username, form_data.password) user = authenticate_user(form_data.username, form_data.password)
if not user: raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail='Incorrect username or password') if not user:
access_token = create_access_token({'sub': user.username, 'role': user.role}, timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)) raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Incorrect username or password")
access_token = create_access_token({"sub": user.username, "role": user.role}, timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES))
return Token(access_token=access_token) return Token(access_token=access_token)
@router.get('/me')
@router.get("/me")
async def me(current_user=Depends(get_current_active_user)): async def me(current_user=Depends(get_current_active_user)):
return {'username': current_user.username, 'role': current_user.role, 'disabled': current_user.disabled} return {"username": current_user.username, "role": current_user.role, "disabled": current_user.disabled}