Open redirect 방지 - 1편

https://example.com?redirect=http://www.iana.org
https://example.com?redirect=https://www.iana.org
https://example.com?redirect=https:\\www.iana.org
https://example.com?redirect=https:/\www.iana.org
https://example.com?redirect=https:\/www.iana.org
https://example.com?redirect=https:\@www.iana.org

# auth.py
async def oauth_login(
    provider: str,
    request: Request,
    redirect_url: str | None = Query(
        None,
        alias="redirect",
        description="로그인 완료 후 돌려보낼 FE URL (허용 오리진만, 미지정·불일치 시 홈으로 폴백)",
    ),
    prompt: str = "",
) -> RedirectResponse:
    """provider(google 등)의 OAuth 인증 페이지로 리다이렉트한다.

    PKCE verifier(+ redirect URL)를 Redis에 저장하고 조회 키(state)를
    httpOnly 쿠키로 내려준다.
    """
    redirect_url = ac.sanitize_redirect_url(redirect_url, _CTX)
    if provider not in settings.OAUTH_PROVIDERS:
        return ac.login_start_failure(
            f"지원하지 않는 프로바이더: {provider}", redirect_url, _CTX
        )

    verifier, challenge = auth_service.generate_pkce()
    state = secrets.token_urlsafe(32)
    try:
        await get_redis().set(
            _CTX.pkce_key.format(state=state),
            json.dumps({"verifier": verifier, "redirect": redirect_url}),
            ex=ac.PKCE_TTL_SECONDS,
        )
    except RedisError:
        return ac.login_start_failure(
            "Redis를 사용할 수 없어 OAuth 로그인을 시작할 수 없습니다", redirect_url, _CTX
        )

    authorize_url = auth_service.build_authorize_url(
        provider, str(request.url_for("oauth_callback")), challenge, prompt
    )
    response = RedirectResponse(authorize_url)
    ac.set_state_cookie(response, state)
    return response

oauth_login()
  └─ sanitize_redirect_url(redirect_url, ctx)   ← 필터링 실행
       ├─ _allowed_redirect_origins(ctx)         ← 화이트리스트 계산
       └─ redirect_home(ctx)                     ← 실패 시 폴백값 생성
  └─ (검증된 값) → Redis 저장
  └─ 실패 시 login_start_failure() → _redirect_with_error()

oauth_callback()
  └─ Redis에서 검증된 값 조회 (재검증 없음)
  └─ 실패 시 callback_failure() → redirect_home() → _redirect_with_error()

def sanitize_redirect_url(redirect_url: str | None, ctx: AuthContext) -> str:
    """FE가 넘긴 redirect URL을 검증한다 — 허용 오리진의 http(s) URL만 통과시키고,
    미지정·불일치·비표준 scheme URL은 에러 대신 홈으로 폴백한다 (open redirect 방지).

    브라우저(WHATWG URL 파서)는 http/https 같은 "special scheme"에서 백슬래시(\\)를
    슬래시(/)와 동일하게 취급해 "https:\\evil.com", "https:/\\evil.com" 같은 입력도
    "<https://evil.com>"으로 정규화해 이동한다. 반면 RFC 3986 기반인 urlsplit은
    백슬래시를 일반 문자로 남겨두므로, 서버와 브라우저가 같은 문자열을 서로 다른
    호스트로 해석할 여지(parser differential)가 생긴다. 지금은 우연히 이런 값이
    urlsplit에서도 netloc이 비어 화이트리스트에 걸리지 않지만, 파서 구현에 기대지
    않고 백슬래시가 섞인 값은 애초에 통째로 거부한다.
    """
    if redirect_url and "\\" not in redirect_url:
        parts = urlsplit(redirect_url)
        origin = f"{parts.scheme}://{parts.netloc}"
        if parts.scheme in ("http", "https") and origin in _allowed_redirect_origins(
            ctx
        ):
            return redirect_url
    return redirect_home(ctx)

def redirect_home(ctx: AuthContext) -> str:
    """로그인 후 돌아갈 기본 FE 홈 — 허용 목록의 첫 URL."""
    urls = ctx.redirect_urls
    return urls[0] if urls else "<http://localhost:3000>"

def _allowed_redirect_origins(ctx: AuthContext) -> set[str]:
    """콜백 후 리다이렉트를 허용할 오리진 — ctx.redirect_urls 기준."""
    origins = set()
    for url in ctx.redirect_urls:
        parts = urlsplit(url)
        origins.add(f"{parts.scheme}://{parts.netloc}")
    return origins

sanitize_redirect_url()
  └─ urlsplit(redirect_url)
       ├─ scheme_chars              ← scheme 파싱 시 문자 검증
       ├─ _UNSAFE_URL_BYTES_TO_REMOVE / _WHATWG_C0_CONTROL_OR_SPACE  ← 전처리
       ├─ _splitnetloc()            ← netloc 분리 (구분자: / ? #)
       └─ _checknetloc()            ← NFKC 동형이의 문자 검사 (ValueError 가능)

scheme_chars = ('abcdefghijklmnopqrstuvwxyz'
                'ABCDEFGHIJKLMNOPQRSTUVWXYZ'
                '0123456789'
                '+-.')

@functools.lru_cache(typed=True)
def urlsplit(url, scheme='', allow_fragments=True):
    url, scheme, _coerce_result = _coerce_args(url, scheme)
    # Only lstrip url as some applications rely on preserving trailing space.
    url = url.lstrip(_WHATWG_C0_CONTROL_OR_SPACE)
    scheme = scheme.strip(_WHATWG_C0_CONTROL_OR_SPACE)

    for b in _UNSAFE_URL_BYTES_TO_REMOVE:
        url = url.replace(b, "")
        scheme = scheme.replace(b, "")

    allow_fragments = bool(allow_fragments)
    netloc = query = fragment = ''
    i = url.find(':')
    if i > 0 and url[0].isascii() and url[0].isalpha():
        for c in url[:i]:
            if c not in scheme_chars:
                break
        else:
            scheme, url = url[:i].lower(), url[i+1:]
    if url[:2] == '//':
        netloc, url = _splitnetloc(url, 2)
        if (('[' in netloc and ']' not in netloc) 
                (  netloc     netloc)):
             ValueError()
           netloc    netloc:
            _check_bracketed_netloc(netloc)
     allow_fragments    url:
        url, fragment = url.split(, )
       url:
        url, query = url.split(, )
    _checknetloc(netloc)
    v = SplitResult(scheme, netloc, url, query, fragment)
     _coerce_result(v)

 ():
    delim = (url)
     c  :    
        wdelim = url.find(c, start)
         wdelim >= :
            delim = (delim, wdelim)
     url[start:delim], url[delim:]

 ():
      netloc  netloc.isascii():
        
    
     unicodedata
    n = netloc.replace(, ).replace(, ).replace(, ).replace(, )
    netloc2 = unicodedata.normalize(, n)
     n == netloc2:
        
     c  :
         c  netloc2:
             ValueError( + netloc +  +
                             )