feat: 포트폴리오·블로그 URL 을 면접 자료로 등록 (US-09)

ai/CLAUDE.md

@@ -93,7 +93,7 @@ ai/
 |-------|-------------|------|
 | `ai.analyze.resume` | `analyze.resume` | 본 구현 (PDF → MD) |
 | `ai.analyze.repository` | `analyze.repository` | 본 구현 (GitHub README + tree + 소스 sampling) |
-| `ai.analyze.web` | `analyze.web` | 본 구현 (URL → trafilatura) |
+| `ai.analyze.web` | `analyze.web` | 본 구현 (URL → trafilatura, SSRF 가드 `analyzer/sources/url_guard.py`) |
 | `ai.analyze.cover_letter` | `analyze.cover_letter` | 본 구현 (자소서 문항 inline 텍스트 → MD, `TextSourceExtractor`) |
 | `ai.generate.questions` | `generate.questions` | 본 구현 (Pro 모델, 질문 풀 생성, US-18) |
 | `ai.generate.followup` | `generate.followup` | 본 구현 (Flash 모델, 답변 평가+꼬리질문, US-19) |
@@ -318,6 +318,12 @@ docker run --env-file .env -p 8000:8000 stackup-ai
 - FastAPI 부트스트랩 + 헬스체크
 - 분석 consumer 본 구현 — `analyze.resume` / `analyze.repository` / `analyze.web`:
   - PDF·GitHub Repo·웹 URL 소스 추출 추상화 (`analyzer/sources/`)
+  - **웹 URL 은 SSRF 가드 필수** (`analyzer/sources/url_guard.py`): 이 프로세스는 docker 네트워크에서
+    Core·PG·RabbitMQ·MinIO 에 닿고 배포 호스트에서는 클라우드 메타데이터(169.254.169.254)에도 닿는다.
+    스킴·userinfo 검사 + 호스트를 **해석한 주소**로 사설/루프백/링크로컬/멀티캐스트/예약/IPv6
+    unique-local 차단. `follow_redirects` 는 끄고 홉마다 재검증(상대 Location 은 절대화 후 검사,
+    5홉 제한). Playwright 렌더 폴백도 검증된 최종 URL 로 실행. Core 의 `WebResumeUrlValidator` 는
+    첫 관문이고 DNS rebinding·리다이렉트로 우회되므로 소켓을 여는 이쪽 검사가 실질 방어선이다.
   - LLM 분석 (`chain/document_analysis_chain.py`, Gemini Pro + Pydantic 출력 파서)
   - 분석 MD를 스토리지에 저장
   - `callback.analysis` 발행 (status `ANALYZED` / `FAILED`, retriable 플래그 포함)

ai/src/ai_server/analyzer/sources/url_guard.py

@@ -0,0 +1,99 @@
+from __future__ import annotations
+
+import ipaddress
+import socket
+from typing import Callable
+from urllib.parse import urlsplit
+
+import structlog
+
+log = structlog.get_logger(__name__)
+
+# host -> IP 문자열 목록. 테스트에서 실제 DNS 를 타지 않도록 주입 가능하게 둔다.
+Resolver = Callable[[str], list[str]]
+
+_ALLOWED_SCHEMES = ("http", "https")
+_REDIRECT_STATUSES = (301, 302, 303, 307, 308)
+
+
+class BlockedUrlError(Exception):
+    """SSRF 가드가 막은 URL."""
+
+    def __init__(self, *, code: str, message: str) -> None:
+        super().__init__(message)
+        self.code = code
+        self.message = message
+
+
+def default_resolver(host: str) -> list[str]:
+    infos = socket.getaddrinfo(host, None, proto=socket.IPPROTO_TCP)
+    return [info[4][0] for info in infos]
+
+
+def is_blocked_address(raw_ip: str) -> bool:
+    """루프백·사설·링크로컬(클라우드 메타데이터)·멀티캐스트·예약·와일드카드 대역인지."""
+    try:
+        ip = ipaddress.ip_address(raw_ip)
+    except ValueError:
+        # 해석할 수 없는 주소는 신뢰하지 않는다.
+        return True
+    return (
+        ip.is_private  # 10/8, 172.16/12, 192.168/16, 127/8, 169.254/16, ::1, fc00::/7
+        or ip.is_loopback
+        or ip.is_link_local
+        or ip.is_multicast
+        or ip.is_reserved
+        or ip.is_unspecified
+    )
+
+
+def assert_public_http_url(url: str, *, resolver: Resolver = default_resolver) -> None:
+    """공개 http(s) 주소가 아니면 :class:`BlockedUrlError`.
+
+    이 프로세스는 docker 네트워크 안에서 Core·PostgreSQL·RabbitMQ·MinIO 에 닿고, 배포
+    호스트에서는 클라우드 메타데이터(169.254.169.254)에도 닿는다. Core 에도 같은 검증이
+    있지만(WebResumeUrlValidator) DNS rebinding·리다이렉트로 우회되므로, **실제 소켓을 여는
+    직전에** 여기서 한 번 더 확인해야 막힌다.
+    """
+    parts = urlsplit(url.strip())
+    scheme = (parts.scheme or "").lower()
+    if scheme not in _ALLOWED_SCHEMES:
+        raise BlockedUrlError(
+            code="INVALID_WEB_URL",
+            message=f"locator must be http(s) URL, got: {url!r}",
+        )
+    # user:pass@host 는 파서 차이를 이용한 호스트 위장에 쓰인다.
+    if parts.username or parts.password:
+        raise BlockedUrlError(
+            code="INVALID_WEB_URL",
+            message="URL 에 사용자 정보를 포함할 수 없음",
+        )
+    host = parts.hostname
+    if not host:
+        raise BlockedUrlError(code="INVALID_WEB_URL", message="URL 에 호스트가 없음")
+
+    try:
+        addresses = resolver(host)
+    except OSError as exc:
+        raise BlockedUrlError(
+            code="WEB_HOST_UNRESOLVED",
+            message=f"호스트를 해석할 수 없음: {host}",
+        ) from exc
+    if not addresses:
+        raise BlockedUrlError(
+            code="WEB_HOST_UNRESOLVED",
+            message=f"호스트를 해석할 수 없음: {host}",
+        )
+
+    for address in addresses:
+        if is_blocked_address(address):
+            # 어떤 내부 주소로 해석됐는지는 로그에만 남긴다.
+            log.warning("web.url.blocked", host=host, resolved=address)
+            raise BlockedUrlError(
+                code="BLOCKED_WEB_URL",
+                message="내부 네트워크 주소는 가져올 수 없음",
+            )
+
+
+def is_redirect(status_code: int) -> bool:
+    return status_code in _REDIRECT_STATUSES

ai/src/ai_server/analyzer/sources/web.py

@@ -7,6 +7,13 @@
 import trafilatura
 
 from ai_server.analyzer.sources.base import ExtractedSource, SourceExtractor
+from ai_server.analyzer.sources.url_guard import (
+    BlockedUrlError,
+    Resolver,
+    assert_public_http_url,
+    default_resolver,
+    is_redirect,
+)
 
 log = structlog.get_logger(__name__)
 
@@ -28,28 +35,27 @@ def __init__(
         max_html_bytes: int = 2_000_000,
         enable_render_fallback: bool = True,
         client: httpx.AsyncClient | None = None,
+        resolver: Resolver = default_resolver,
+        max_redirects: int = 5,
     ) -> None:
         self._timeout_sec = timeout_sec
         self._max_html_bytes = max_html_bytes
         self._enable_render_fallback = enable_render_fallback
         self._client = client
+        self._resolver = resolver
+        self._max_redirects = max_redirects
 
     async def extract(self, locator: str) -> ExtractedSource:
         url = locator.strip()
-        if not (url.startswith("http://") or url.startswith("https://")):
-            raise WebFetchError(
-                code="INVALID_WEB_URL",
-                message=f"locator must be http(s) URL, got: {locator!r}",
-                retriable=False,
-            )
+        self._require_public(url)
 
         html, final_url, content_type = await self._fetch_html(url)
         text = await asyncio.to_thread(_extract_main_text, html, final_url)
         rendered = False
 
         # 본문이 비면 JS 렌더링 SPA(React 포폴 등)일 가능성 → Playwright 로 렌더 후 재추출.
         if not text.strip() and self._enable_render_fallback:
-            rendered_html = await self._render(url)
+            rendered_html = await self._render(final_url)
             if rendered_html:
                 html = rendered_html
                 text = await asyncio.to_thread(_extract_main_text, html, final_url)
@@ -100,12 +106,26 @@ async def _render(self, url: str) -> str | None:
             log.warning("web.render.failed", url=url, error=str(exc))
             return None
 
+    # SSRF 가드. Core 에서도 검증하지만 DNS rebinding·리다이렉트로 우회되므로
+    # 실제 요청을 보내기 직전에(그리고 리다이렉트 홉마다) 다시 확인한다.
+    def _require_public(self, url: str) -> None:
+        try:
+            assert_public_http_url(url, resolver=self._resolver)
+        except BlockedUrlError as err:
+            raise WebFetchError(
+                code=err.code,
+                message=err.message,
+                retriable=False,
+            ) from err
+
     async def _fetch_html(self, url: str) -> tuple[str, str, str]:
         if self._client is not None:
             return await self._do_fetch(self._client, url)
         async with httpx.AsyncClient(
             timeout=self._timeout_sec,
-            follow_redirects=True,
+            # 자동 추적을 끄고 홉마다 목적지를 검증한다 — 공개 URL 이 내부 주소로
+            # 리다이렉트하는 경로를 막기 위해.
+            follow_redirects=False,
             headers={
                 "User-Agent": "StackUp-AI/1.0 (+resume web extractor)",
                 "Accept": "text/html,application/xhtml+xml",
@@ -118,14 +138,29 @@ async def _do_fetch(
         client: httpx.AsyncClient,
         url: str,
     ) -> tuple[str, str, str]:
-        try:
-            resp = await client.get(url)
-        except httpx.HTTPError as exc:
+        current = url
+        for _ in range(self._max_redirects + 1):
+            try:
+                resp = await client.get(current)
+            except httpx.HTTPError as exc:
+                raise WebFetchError(
+                    code="WEB_FETCH_FAILED",
+                    message=f"HTTP 요청 실패: {exc}",
+                    retriable=True,
+                ) from exc
+
+            location = resp.headers.get("location")
+            if not (is_redirect(resp.status_code) and location):
+                break
+            # 상대 Location 도 절대 URL 로 만든 뒤 검증한다.
+            current = str(httpx.URL(current).join(location))
+            self._require_public(current)
+        else:
             raise WebFetchError(
-                code="WEB_FETCH_FAILED",
-                message=f"HTTP 요청 실패: {exc}",
-                retriable=True,
-            ) from exc
+                code="WEB_TOO_MANY_REDIRECTS",
+                message=f"리다이렉트가 한도({self._max_redirects})를 초과",
+                retriable=False,
+            )
 
         if resp.status_code >= 400:
             raise WebFetchError(

ai/tests/test_web_extractor.py

@@ -6,6 +6,13 @@
 from ai_server.analyzer.sources.web import WebFetchError, WebSourceExtractor
 
 
+# DNS 를 타지 않는다 — example.com 계열은 공개 IP, 그 외 리터럴은 그대로.
+def _fake_resolver(host: str) -> list[str]:
+    if host.endswith("example.com"):
+        return ["93.184.216.34"]
+    return [host]
+
+
 def _make_client(
     *,
     status: int = 200,
@@ -39,7 +46,7 @@ async def test_extract_returns_main_body_text() -> None:
         "<footer>foot</footer></body></html>"
     )
     client = _make_client(body=html)
-    extractor = WebSourceExtractor(client=client)
+    extractor = WebSourceExtractor(client=client, resolver=_fake_resolver)
     result = await extractor.extract("https://example.com/r")
 
     assert result.source_type == "WEB"
@@ -51,7 +58,7 @@ async def test_extract_returns_main_body_text() -> None:
 
 @pytest.mark.asyncio
 async def test_rejects_non_http_locator() -> None:
-    extractor = WebSourceExtractor(client=_make_client())
+    extractor = WebSourceExtractor(client=_make_client(), resolver=_fake_resolver)
     with pytest.raises(WebFetchError) as exc_info:
         await extractor.extract("ftp://example.com/x")
     assert exc_info.value.code == "INVALID_WEB_URL"
@@ -61,7 +68,7 @@ async def test_rejects_non_http_locator() -> None:
 @pytest.mark.asyncio
 async def test_raises_on_http_error_status() -> None:
     client = _make_client(status=503)
-    extractor = WebSourceExtractor(client=client)
+    extractor = WebSourceExtractor(client=client, resolver=_fake_resolver)
     with pytest.raises(WebFetchError) as exc_info:
         await extractor.extract("https://example.com/r")
     assert exc_info.value.code == "WEB_HTTP_STATUS"
@@ -71,7 +78,7 @@ async def test_raises_on_http_error_status() -> None:
 @pytest.mark.asyncio
 async def test_raises_on_4xx_as_non_retriable() -> None:
     client = _make_client(status=404)
-    extractor = WebSourceExtractor(client=client)
+    extractor = WebSourceExtractor(client=client, resolver=_fake_resolver)
     with pytest.raises(WebFetchError) as exc_info:
         await extractor.extract("https://example.com/r")
     assert exc_info.value.code == "WEB_HTTP_STATUS"
@@ -81,7 +88,7 @@ async def test_raises_on_4xx_as_non_retriable() -> None:
 @pytest.mark.asyncio
 async def test_rejects_non_html_content_type() -> None:
     client = _make_client(content_type="application/pdf")
-    extractor = WebSourceExtractor(client=client)
+    extractor = WebSourceExtractor(client=client, resolver=_fake_resolver)
     with pytest.raises(WebFetchError) as exc_info:
         await extractor.extract("https://example.com/r")
     assert exc_info.value.code == "WEB_NOT_HTML"
@@ -91,7 +98,9 @@ async def test_rejects_non_html_content_type() -> None:
 async def test_rejects_oversized_html() -> None:
     big = b"<html>" + b"a" * 1024 + b"</html>"
     client = _make_client(body=big)
-    extractor = WebSourceExtractor(client=client, max_html_bytes=512)
+    extractor = WebSourceExtractor(
+        client=client, max_html_bytes=512, resolver=_fake_resolver
+    )
     with pytest.raises(WebFetchError) as exc_info:
         await extractor.extract("https://example.com/r")
     assert exc_info.value.code == "WEB_HTML_TOO_LARGE"
@@ -100,7 +109,9 @@ async def test_rejects_oversized_html() -> None:
 @pytest.mark.asyncio
 async def test_raises_on_empty_body() -> None:
     client = _make_client(body="<html><body></body></html>")
-    extractor = WebSourceExtractor(client=client, enable_render_fallback=False)
+    extractor = WebSourceExtractor(
+        client=client, enable_render_fallback=False, resolver=_fake_resolver
+    )
     with pytest.raises(WebFetchError) as exc_info:
         await extractor.extract("https://example.com/r")
     assert exc_info.value.code == "EMPTY_WEB_BODY"
@@ -111,7 +122,7 @@ async def test_raises_on_empty_body() -> None:
 async def test_empty_body_falls_back_to_render() -> None:
     # 1차 fetch = JS 셸(본문 없음) → 렌더 폴백으로 본문 확보
     client = _make_client(body='<html><body><div id="root"></div></body></html>')
-    extractor = WebSourceExtractor(client=client)
+    extractor = WebSourceExtractor(client=client, resolver=_fake_resolver)
     rendered_html = (
         "<html><body><article><h1>김OO</h1>"
         "<p>프론트엔드 개발자. React 포트폴리오.</p></article></body></html>"
@@ -127,7 +138,7 @@ async def test_empty_body_falls_back_to_render() -> None:
 @pytest.mark.asyncio
 async def test_render_fallback_returning_none_raises_empty() -> None:
     client = _make_client(body='<html><body><div id="root"></div></body></html>')
-    extractor = WebSourceExtractor(client=client)
+    extractor = WebSourceExtractor(client=client, resolver=_fake_resolver)
     extractor._render = AsyncMock(return_value=None)  # 렌더 실패/불가
     with pytest.raises(WebFetchError) as exc_info:
         await extractor.extract("https://example.com/spa")
@@ -137,7 +148,7 @@ async def test_render_fallback_returning_none_raises_empty() -> None:
 @pytest.mark.asyncio
 async def test_raises_on_httpx_error_as_retriable() -> None:
     client = _make_client(raise_exc=httpx.ConnectError("dns fail"))
-    extractor = WebSourceExtractor(client=client)
+    extractor = WebSourceExtractor(client=client, resolver=_fake_resolver)
     with pytest.raises(WebFetchError) as exc_info:
         await extractor.extract("https://example.com/r")
     assert exc_info.value.code == "WEB_FETCH_FAILED"