55 lines
1.6 KiB
Python
55 lines
1.6 KiB
Python
|
import uvicorn
|
||
|
|
from passlib.context import CryptContext
|
||
|
|
from datetime import datetime, timedelta
|
||
|
|
import jwt
|
||
|
|
from fastapi import FastAPI, HTTPException, Request
|
||
|
|
from models.User import User
|
||
|
|
from controllers.JWT import secret_key
|
||
|
|
|
||
|
|
|
||
|
|
app = FastAPI()
|
||
|
|
pwd_context = CryptContext(schemes=["bcrypt"])
|
||
|
|
# 模拟数据库中的用户
|
||
|
|
|
||
|
|
users_db = {
|
||
|
|
"admin": {
|
||
|
|
"username": "admin",
|
||
|
|
"password": pwd_context.hash("admin123")
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
@app.post("/login")
|
||
|
|
async def login(user: User):
|
||
|
|
if user.username not in users_db:
|
||
|
|
raise HTTPException(status_code=401, detail="Invalid username")
|
||
|
|
stored_user = users_db[user.username]
|
||
|
|
if not pwd_context.verify(user.password, stored_user["password"]):
|
||
|
|
raise HTTPException(status_code=401, detail="Invalid password")
|
||
|
|
token = await generate_token(user.username)
|
||
|
|
return {"access_token": token}
|
||
|
|
|
||
|
|
|
||
|
|
async def generate_token(username: str) -> str:
|
||
|
|
expiration = datetime.utcnow() + timedelta(minutes=30)
|
||
|
|
payload = {"username": username, "exp": expiration}
|
||
|
|
return jwt.encode(payload, secret_key, algorithm="HS256")
|
||
|
|
|
||
|
|
|
||
|
|
@app.get("/users/me")
|
||
|
|
async def get_user_profile(request: Request):
|
||
|
|
cookie = request.headers.get("Cookie")
|
||
|
|
try:
|
||
|
|
payload = jwt.decode(cookie, secret_key, algorithms=["HS256"])
|
||
|
|
username = payload["username"]
|
||
|
|
if username not in users_db:
|
||
|
|
raise HTTPException(status_code=401, detail="Invalid username")
|
||
|
|
return {"username": username}
|
||
|
|
except jwt.DecodeError:
|
||
|
|
raise HTTPException(status_code=401, detail="Invalid token")
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == '__main__':
|
||
|
|
uvicorn.run(app='main:app', host="0.0.0.0", port=80, reload=True)
|
||
|
|
|