OAuth2 스코프

OAuth2 스코프 (OAuth2 Scopes)

FastAPI에서 OAuth2 스코프를 바로 사용할 수 있어요. 매끄럽게 통합되어 있죠.

이렇게 하면 OAuth2 표준을 따라, OpenAPI 앱(그리고 API 문서)에 통합된 더 세밀한 권한 시스템을 가질 수 있어요.

스코프가 있는 OAuth2는 페이스북, 구글, GitHub, 마이크로소프트, X(트위터) 등 많은 대형 인증 제공자들이 쓰는 메커니즘이에요. 사용자와 앱에 특정 권한을 부여하는 데 사용합니다.

여러분이 페이스북·구글·GitHub·마이크로소프트·X로 "로그인할" 때마다, 그 앱은 스코프가 있는 OAuth2를 사용하고 있는 거예요.

이 섹션에서는 FastAPI 앱에서 같은 OAuth2 with scopes로 인증과 권한 부여를 다루는 방법을 살펴볼게요.

경고: 이건 어느 정도 고급 섹션이에요. 이제 막 시작했다면 건너뛰어도 됩니다. OAuth2 스코프가 꼭 필요한 건 아니고, 원하는 대로 인증·인가를 다룰 수 있어요. 다만 스코프가 있는 OAuth2는 API(OpenAPI 포함)와 API 문서에 잘 통합될 수 있어요. 그래도 스코프나 다른 보안·인가 요구사항을 적용하는 건 여전히 코드에서 원하는 대로 해야 해요. 많은 경우 OAuth2 with scopes는 과할 수 있어요. 하지만 필요하다는 걸 알거나 궁금하다면, 계속 읽어 보세요.

출처: 공식문서

OAuth2 스코프와 OpenAPI

OAuth2 명세는 "스코프(scopes)"를 공백으로 구분된 문자열 목록으로 정의해요.

각 문자열의 내용은 어떤 형식이든 될 수 있지만, 공백을 포함하면 안 됩니다.

이 스코프들은 "권한(permissions)"을 나타냅니다.

OpenAPI(예: API 문서)에서는 "security schemes"를 정의할 수 있어요.

이 security scheme 중 하나가 OAuth2를 사용하면, 스코프도 선언하고 사용할 수 있습니다.

각 "스코프"는 그냥 문자열(공백 없음)이에요. 보통 특정 보안 권한을 선언하는 데 사용됩니다. 예를 들어:

• users:read나 users:write는 흔한 예시예요.
• instagram_basic은 페이스북 / 인스타그램에서 쓰고요.
• https://www.googleapis.com/auth/drive는 구글이 사용해요.

참고: OAuth2에서 "스코프"는 필요한 특정 권한을 선언하는 문자열일 뿐이에요. : 같은 다른 문자가 들어 있거나 URL이어도 상관없어요. 그런 세부사항은 구현별로 달라요. OAuth2한테는 그냥 문자열일 뿐입니다.

전체 그림 (Global view)

먼저, 메인 Tutorial - User Guide의 OAuth2 with Password (and hashing), Bearer with JWT tokens 예시에서 바뀌는 부분들을 스코프를 쓰지 않고 빠르게 확인해 볼게요. 이제 OAuth2 스코프를 써서요:

from datetime import datetime, timedelta, timezone
from typing import Annotated

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: str | None = None
    scopes: list[str] = []


class User(BaseModel):
    username: str
    email: str | None = None
    full_name: str | None = None
    disabled: bool | None = None


class UserInDB(User):
    hashed_password: str


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."},
)

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:
        verify_password(password, DUMMY_HASH)
        return False
    if not verify_password(password, user.hashed_password):
        return False
    return user


def create_access_token(data: dict, expires_delta: 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/")
async def read_users_me(
    current_user: Annotated[User, Depends(get_current_active_user)],
) -> 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"}

🤓 다른 버전과 변형

팁: 가능하면 Annotated 버전을 사용하세요.

from datetime import datetime, timedelta, timezone

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: str | None = None
    scopes: list[str] = []


class User(BaseModel):
    username: str
    email: str | None = None
    full_name: str | None = None
    disabled: bool | None = None


class UserInDB(User):
    hashed_password: str


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."},
)

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:
        verify_password(password, DUMMY_HASH)
        return False
    if not verify_password(password, user.hashed_password):
        return False
    return user


def create_access_token(data: dict, expires_delta: 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/")
async def read_users_me(current_user: User = Depends(get_current_active_user)) -> 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"}

이제 이 변경들을 단계별로 살펴볼게요.

OAuth2 Security scheme

첫 번째 변경은, 이제 OAuth2 security scheme을 me와 items 두 개의 사용 가능한 스코프와 함께 선언한다는 거예요.

scopes 파라미터는 각 스코프를 키로, 설명을 값으로 하는 dict를 받아요:

oauth2_scheme = OAuth2PasswordBearer(
    tokenUrl="token",
    scopes={"me": "Read information about the current user.", "items": "Read items."},
)

이제 스코프를 선언했기 때문에, 로그인/인가할 때 API 문서에 나타납니다.

그리고 어떤 스코프에 접근 권한을 줄지 선택할 수 있어요: me와 items 말이죠.

이건 페이스북·구글·GitHub 등으로 로그인하면서 권한을 줄 때 쓰는 것과 같은 메커니즘이에요.

스코프가 있는 JWT 토큰

이제 token _path operation_이 요청된 스코프를 돌려주도록 수정해 볼게요.

여전히 같은 OAuth2PasswordRequestForm을 사용해요. 여기에는 요청에서 받은 각 스코프를 담는 list of str 속성 scopes가 있습니다.

그리고 그 스코프를 JWT 토큰의 일부로 돌려줍니다.

위험: 간단하게 하기 위해, 여기서는 받은 스코프를 그대로 토큰에 추가하고 있어요. 하지만 실제 앱에서는 보안을 위해, 사용자가 실제로 가질 수 있는 스코프나 미리 정의한 스코프만 추가하도록 확실히 해야 해요.

# 로그인 시 요청받은 스코프를 토큰에 담아 돌려주는 부분
@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")

_path operations_과 의존성에 스코프 선언하기

이제 /users/me/items/의 _path operation_이 items 스코프를 요구한다고 선언합니다.

이를 위해 fastapi에서 Security를 import 해서 사용해요.

Security로 의존성을 선언할 수 있어요(Depends처럼요). 하지만 Security는 스코프(문자열) 목록인 scopes 파라미터도 받습니다.

이 경우 Security에 의존성 함수 get_current_active_user를 전달해요(Depends로 할 때와 같은 방식이죠).

하지만 스코프의 list도 전달하는데, 이 경우엔 스코프 하나뿐이에요: items(더 많을 수도 있어요).

그리고 의존성 함수 get_current_active_user는 자신의 하위 의존성을 Depends만이 아니라 Security로도 선언할 수 있어요. 자신의 하위 의존성 함수(get_current_user)와 추가 스코프 요구사항을 선언하죠.

이 경우 me 스코프를 요구합니다(둘 이상을 요구할 수도 있어요).

참고: 반드시 여러 곳에 다른 스코프를 추가할 필요는 없어요. 여기서는 FastAPI가 다른 수준에서 선언된 스코프를 어떻게 다루는지 보여주기 위해 그렇게 하는 거예요.

# items 스코프와 me 스코프를 요구하는 예시
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.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}]

기술적 세부사항: Security는 사실 Depends의 하위 클래스이고, 나중에 볼 추가 파라미터 하나만 더 있어요. Depends 대신 Security를 쓰면 FastAPI는 보안 스코프를 선언할 수 있고, 내부적으로 사용하며 OpenAPI로 API를 문서화한다는 걸 알게 됩니다. fastapi에서 Query, Path, Depends, Security 등을 import 하면, 사실 그건 특별한 클래스를 돌려주는 함수들이에요.

SecurityScopes 사용하기

이제 의존성 get_current_user를 갱신해 볼게요.

이건 위 의존성들이 사용하는 것이에요.

여기서는 이전에 만든 같은 OAuth2 scheme을 의존성으로 선언해서 사용합니다: oauth2_scheme 말이죠.

이 의존성 함수 자체에는 스코프 요구사항이 없으므로, oauth2_scheme에는 Depends를 쓸 수 있어요. 보안 스코프를 지정할 필요가 없을 때는 Security를 쓸 필요가 없습니다.

또한 fastapi.security에서 import 한 SecurityScopes 타입의 특별한 파라미터를 선언해요.

이 SecurityScopes 클래스는 Request와 비슷합니다(Request는 request 객체를 직접 얻는 데 쓰였죠).

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

SecurityScopes로 "security scopes"를 주입받고, security_scopes.scopes에 요구되는 스코프 목록이 들어 있어요. get_current_user 함수는 토큰 안의 scope 클레임을 공백으로 나눠 token_data.scopes에 담고, 요구 스코프가 토큰 스코프에 포함되어 있는지 하나씩 확인해서, 없으면 "Not enough permissions"라는 401 예외를 던집니다.

더 알아보기 (Learn more)

• 공식문서: OAuth2 Scopes