로그인 추가하기

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from datetime import datetime, timedelta
from typing import Optional
import uuid, hashlib, base64, hmac, requests

app = FastAPI()

# Secret for user JWT token
JWT_SECRET = "your-jwt-secret-key"
JWT_ALGORITHM = "HS256"
JWT_EXPIRE_MINUTES = 60

# Dummy user
fake_user = {
    "username": "abc",
    "password": "ert"  # Plaintext for simplicity (use hashed passwords in prod!)
}

# OAuth2 scheme
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login")

SERVER_URL = "https://test.com"

# === Auth ===
def create_access_token(data: dict, expires_delta: timedelta):
    to_encode = data.copy()
    expire = datetime.utcnow() + expires_delta
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, JWT_SECRET, algorithm=JWT_ALGORITHM)

def verify_token(token: str = Depends(oauth2_scheme)):
    try:
        payload = jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALGORITHM])
        username = payload.get("sub")
        if username != fake_user["username"]:
            raise HTTPException(status_code=401, detail="Invalid user")
        return username
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

@app.post("/login")
def login(form_data: OAuth2PasswordRequestForm = Depends()):
    if form_data.username != fake_user["username"] or form_data.password != fake_user["password"]:
        raise HTTPException(status_code=400, detail="Incorrect username or password")

    access_token = create_access_token(
        data={"sub": fake_user["username"]},
        expires_delta=timedelta(minutes=JWT_EXPIRE_MINUTES)
    )
    return {"access_token": access_token, "token_type": "bearer"}

# === Upbit Request Helper ===
def generate_jwt(query_string: str) -> str:
    query_hash = hashlib.sha512(query_string.encode()).hexdigest()

    header = {"alg": "HS256", "typ": "JWT"}
    payload = {
        "access_key": ACCESS_KEY,
        "nonce": str(uuid.uuid4()),
        "query_hash": query_hash,
        "query_hash_alg": "SHA512"
    }

    def b64url(d):
        return base64.urlsafe_b64encode(str(d).replace("'", '"').encode()).decode().rstrip("=")

    h_b64 = b64url(header)
    p_b64 = b64url(payload)
    sig = hmac.new(SECRET_KEY.encode(), f"{h_b64}.{p_b64}".encode(), hashlib.sha256).digest()
    s_b64 = base64.urlsafe_b64encode(sig).decode().rstrip("=")

    return f"{h_b64}.{p_b64}.{s_b64}"

# === Protected Route ===
@app.get("/test")
def get(
    state: str = "done",
    page: int = 1,
    user: str = Depends(verify_token)
):
    params = {
        "state": state,
        "page": page,
    }

    query_string = "&".join(f"{k}={v}" for k, v in params.items())
    jwt_token = generate_jwt(query_string)

    headers = {
        "Authorization": f"Bearer {jwt_token}"
    }

    response = requests.get(f"{SERVER_URL}/v1/orders", headers=headers, params=params)

    if response.ok:
        return response.json()
    else:
        return {"error": response.status_code, "message": response.text}

Token 얻기

curl -X POST http://localhost:8000/login \
  -d "username=abc&password=ert" \
  -H "Content-Type: application/x-www-form-urlencoded"

/test 호출

curl http://localhost:8000/test \
  -H "Authorization: Bearer <your_token_here>"