From ce4b418603ca971ef835d8da6a85693ea54bca76 Mon Sep 17 00:00:00 2001 From: devmrko Date: Thu, 23 Jul 2026 14:40:47 +0900 Subject: [PATCH] refs #709: replace portal query token with HttpOnly auth --- .../709-hmm-portal-http-only-auth/README.md | 125 ++++ ...-23-hmm-portal-cookie-auth-verification.md | 80 +++ .../apps/poc4/mcp_discovery_ui.py | 166 +----- .../nginx/hmm-portal-auth.locations.conf | 41 ++ .../infra/systemd/hmm-portal-auth.service | 29 + .../src/agent_console/auth_gateway.py | 549 ++++++++++++++++++ .../src/agent_console/presentation.py | 6 + .../tests/test_auth_gateway.py | 217 +++++++ 8 files changed, 1079 insertions(+), 134 deletions(-) create mode 100644 docs/design/709-hmm-portal-http-only-auth/README.md create mode 100644 docs/reports/2026-07-23-hmm-portal-cookie-auth-verification.md create mode 100644 poc4_active_source_20260714/infra/nginx/hmm-portal-auth.locations.conf create mode 100644 poc4_active_source_20260714/infra/systemd/hmm-portal-auth.service create mode 100644 poc4_active_source_20260714/src/agent_console/auth_gateway.py create mode 100644 poc4_active_source_20260714/tests/test_auth_gateway.py diff --git a/docs/design/709-hmm-portal-http-only-auth/README.md b/docs/design/709-hmm-portal-http-only-auth/README.md new file mode 100644 index 0000000..f59d470 --- /dev/null +++ b/docs/design/709-hmm-portal-http-only-auth/README.md @@ -0,0 +1,125 @@ +# HMM 포털 URL 토큰 제거와 HttpOnly 쿠키 인증 설계 (#709) + +> 상태: 구현·배포·검증 완료 +> 대상: `https://hmm.cloud-handson.com` +> 브랜치: `hmm-backoffice` + +## 문제 + +현재 Streamlit 로그인 유지 기능은 서명된 토큰을 `poc4_remember` query parameter에 저장한다. +토큰이 암호화된 비밀번호는 아니더라도 유효 기간 동안 인증 수단으로 작동하므로 다음 위치에 남을 수 +있다. + +- 브라우저 주소와 방문 기록 +- Nginx·상위 프록시 access log +- 사용자가 복사한 링크와 화면 캡처 +- 외부 링크 이동 시 Referer + +인증 수단은 URL에 포함하지 않는다. 기존 query-token 코드를 삭제하고 기존 서명 secret을 +회전해 과거 URL을 즉시 무효화한다. + +## 목표 + +- 로그인은 `POST /auth/login`으로만 처리한다. +- 인증 상태는 `Secure`, `HttpOnly`, `SameSite=Lax`, `Path=/` 쿠키에만 둔다. +- Nginx가 모든 Streamlit HTTP·WebSocket 요청 전에 쿠키를 검증한다. +- Streamlit은 외부 요청 헤더가 아니라 Nginx가 덮어쓴 내부 사용자 헤더만 사용한다. +- 로그아웃은 쿠키를 만료시키고 로그인 화면으로 돌아간다. +- 로그인·로그아웃 이후 주소에 토큰이나 자격 증명이 남지 않는다. + +## 구성 + +```text +Browser + ├─ GET /auth/login ────────────────┐ + ├─ POST /auth/login (ID/password) │ + └─ Cookie: __Host-HMM_PORTAL_SESSION + ▼ +Nginx :443 + ├─ /auth/* ───────────────► auth_gateway.py :8621 + └─ /* + auth_request ──────► /auth/check + ├─ 204 + X-Auth-User ─► Streamlit :8622 + └─ 401 ───────────────► /auth/login +``` + +인증 서비스와 Streamlit은 모두 `127.0.0.1`에만 바인딩한다. 외부에서 인증 사용자 헤더를 +보내더라도 Nginx가 `auth_request` 결과로 값을 덮어쓴다. + +## 쿠키 + +| 항목 | 값 | +|---|---| +| 이름 | `__Host-HMM_PORTAL_SESSION` | +| 속성 | `Secure; HttpOnly; SameSite=Lax; Path=/` | +| 기본 로그인 | 브라우저 세션 쿠키, 서버 토큰 만료 12시간 | +| 로그인 유지 | `Max-Age=604800`, 서버 토큰 만료 7일 | +| 형식 | version, user, issued-at, expiry, nonce를 담은 base64url payload + HMAC-SHA256 | +| 서명키 | `POC4_LOGIN_COOKIE_SECRET`, Git·로그 미기록 | + +`__Host-` 접두사는 `Secure`, `Path=/`, Domain 미지정 조건을 강제해 하위 도메인의 쿠키 +주입 범위를 줄인다. + +## 로그인 보호 + +- PBKDF2 비밀번호 해시는 기존 `POC4_LOGIN_PASSWORD_PBKDF2`를 사용한다. +- 로그인 GET에서 10분 유효한 일회용 CSRF 쿠키와 hidden 값을 발급한다. +- 로그인 POST는 CSRF 두 값을 상수 시간 비교한 뒤 자격 증명을 확인한다. +- 오류 메시지는 사용자 존재 여부와 비밀번호 실패를 구분하지 않는다. +- 요청 body와 필드 길이를 제한한다. +- 실패 횟수는 IP별 짧은 시간 창에서 제한한다. +- 인증 응답에는 `Cache-Control: no-store`와 보안 헤더를 설정한다. +- 서비스 로그에는 query string, 쿠키, 비밀번호, 토큰을 기록하지 않는다. + +## Streamlit 변경 + +- `poc4_remember` 상수, 생성, 복원, query 정리 코드를 삭제한다. +- Streamlit 내부 로그인 유지 로직을 삭제한다. +- `st.context.headers["X-HMM-Authenticated-User"]`가 설정된 경우에만 포털 세션을 활성화한다. +- 기대 사용자와 프록시 사용자 값은 상수 시간 비교한다. +- 로그아웃 UI는 `/auth/logout`으로 이동해 쿠키를 만료시킨다. +- 신뢰 헤더가 없으면 자격 증명 폼 대신 인증 게이트웨이 설정 오류만 표시한다. + +## 배포 + +1. 인증 서비스 소스와 systemd unit을 `/opt/hmm-poc4`에 배포한다. +2. 새 `POC4_LOGIN_COOKIE_SECRET`을 root 소유 환경 파일에 추가하고 기존 + `POC4_LOGIN_REMEMBER_SECRET`은 제거한다. +3. 인증 서비스를 `127.0.0.1:8621`에서 시작한다. +4. Nginx 설정에 `/auth/*`, 내부 `/auth/check`, `auth_request`를 적용한다. +5. Streamlit 소스를 배포하고 서비스를 재시작한다. +6. `nginx -t`, 서비스 상태, 로그인·쿠키·WebSocket·로그아웃을 검증한다. + +## 완료 검증 + +- 기존 `?poc4_remember=` 요청이 인증되지 않고 로그인 화면으로 이동한다. +- 로그인 POST 응답의 `Location`은 `/`이고 URL에 토큰이 없다. +- 세션 쿠키에 `Secure`, `HttpOnly`, `SameSite=Lax`, `Path=/`가 모두 있다. +- 조작·만료 쿠키는 `/auth/check`에서 401이다. +- 인증 쿠키가 없으면 Streamlit asset·WebSocket을 포함한 보호 경로를 사용할 수 없다. +- 로그인 후 포털 주요 탭, MCP 설정, 사용자 전환이 정상 동작한다. +- 로그아웃 후 쿠키가 만료되고 보호 경로가 다시 로그인 화면으로 이동한다. + +## 롤백 + +변경 전 Nginx 설정, Streamlit 소스, 환경 파일을 타임스탬프 백업한다. 장애 시 이 세 파일을 +복구하고 인증 서비스를 중지한다. 롤백을 해도 query-token 구현은 재활성화하지 않으며, 임시로 +포털 접근을 차단하는 쪽을 우선한다. + +## 배포 검증 결과 + +2026-07-23 운영 배포에서 다음을 확인했다. + +- `hmm-portal-auth.service`, `poc4-streamlit.service`, `nginx` 모두 `active` +- 기존 query-token 서명키 제거·회전, 환경 백업의 이전 서명키도 제거 +- `/` 미인증 요청: `/auth/login`으로 이동 +- `/?poc4_remember=retired-token`: 인증되지 않고 `/auth/login`으로 이동하며 query 제거 +- 로그인 페이지: URL token 없음, CSRF cookie는 `Secure; HttpOnly; SameSite=Strict` +- 포털 session cookie: `Secure; HttpOnly; SameSite=Lax; Path=/` +- 브라우저 `document.cookie`에서 session cookie를 읽을 수 없음 +- 인증 후 URL: `https://hmm.cloud-handson.com/`, query 없음 +- 아키텍처·시나리오·감사로그·보안관리 탭 및 MCP endpoint 설정 표시 정상 +- 로그아웃 후 session cookie 제거와 로그인 화면 복귀 확인 +- Python 단위·HTTP 통합 테스트 21건 통과, 선택적 Streamlit runtime 테스트 1건 skip +- 브라우저 page error 0건, console error 0건 + +상세 증거는 `docs/reports/2026-07-23-hmm-portal-cookie-auth-verification.md`에 기록한다. diff --git a/docs/reports/2026-07-23-hmm-portal-cookie-auth-verification.md b/docs/reports/2026-07-23-hmm-portal-cookie-auth-verification.md new file mode 100644 index 0000000..0bb6342 --- /dev/null +++ b/docs/reports/2026-07-23-hmm-portal-cookie-auth-verification.md @@ -0,0 +1,80 @@ +# HMM 포털 HttpOnly 쿠키 인증 적용·검증 보고서 + +- 일자: 2026-07-23 +- Redmine: #709 +- 브랜치: `hmm-backoffice` +- 서비스: `https://hmm.cloud-handson.com` + +## 수정 결과 + +Streamlit의 `poc4_remember` query-token 생성·복원 코드를 삭제했다. Nginx가 모든 포털 +HTTP·WebSocket 요청에 `auth_request`를 수행하고, localhost 인증 서비스가 검증한 사용자와 +만료 시각만 Streamlit에 전달한다. + +| 구성 | 결과 | +|---|---| +| 인증 서비스 | `hmm-portal-auth.service` / active | +| 인증 서비스 bind | `127.0.0.1:8621` | +| 포털 | `poc4-streamlit.service` / active | +| 공개 경계 | Nginx `auth_request` | +| session cookie | `__Host-HMM_PORTAL_SESSION` | +| cookie 속성 | `Secure; HttpOnly; SameSite=Lax; Path=/` | +| 로그인 CSRF | 10분 일회용 double-submit cookie | +| URL token | 제거 | +| 이전 서명키 | 운영 환경과 환경 백업에서 제거 | + +## 자동 테스트 + +```text +python3 -m unittest discover -s tests -p 'test_*.py' -v +Ran 21 tests +OK (skipped=1) +``` + +검증 항목: + +- PBKDF2 비밀번호 비교 +- session token 발급·검증·만료·조작 거부 +- persistent/session/logout cookie 속성 +- login rate limit +- 실제 HTTP login → auth check → tamper reject → logout 흐름 +- login 성공 redirect가 `/`이고 token·remember query가 없는지 확인 + +## 운영 HTTP 검증 + +| 요청 | 결과 | +|---|---| +| 쿠키 없이 `/` | 302 → `/auth/login` | +| `/?poc4_remember=retired-token` | 302 → `/auth/login`, query 전달 안 됨 | +| `/auth/login` | 200, `Cache-Control: no-store` | +| 조작 session cookie | 401 | +| 유효 session cookie | Streamlit 200 | +| `/auth/logout` | session cookie `Max-Age=0`, 로그인 화면 이동 | + +로그인 페이지 응답에는 `X-Content-Type-Options`, `X-Frame-Options`, `Referrer-Policy: +no-referrer`, 제한된 CSP가 포함된다. + +## 실제 브라우저 검증 + +Playwright에서 다음을 확인했다. + +- 미인증 query-token URL은 `/auth/login`으로 이동하고 최종 URL에서 query가 사라짐 +- 로그인 화면의 사용자 ID, 비밀번호, 로그인 유지 UI 표시 +- 인증 후 최종 URL은 `/`, query 없음 +- `__Host-HMM_PORTAL_SESSION`: HttpOnly=true, Secure=true, SameSite=Lax, Path=/ +- `document.cookie`에 portal session cookie가 없음 +- 아키텍처·시나리오·감사로그·보안관리 탭 필수 내용 표시 +- `https://hmm-mcp.cloud-handson.com/mcp` 설정 표시 +- 로그아웃 후 session cookie 없음 +- page error 0, console error 0 + +캡처와 기계 판독 보고서는 운영 검증 작업 디렉터리 +`/private/tmp/hmm-cookie-auth-audit/`에 생성했다. + +## 보안 정리 + +- 노출된 과거 query-token은 새 인증 경로에서 사용되지 않으며 기존 HMAC 서명키도 회전했다. +- 회전 전 환경 백업 두 개에서는 이전·중간 서명키 줄을 제거했다. +- 현재 secret은 `/opt/hmm-poc4/.env`에만 있고 파일 권한은 `opc:opc 0600`이다. +- 토큰, 비밀번호, cookie 값은 Git·Redmine·보고서·서비스 로그에 기록하지 않았다. +- Nginx access log에는 앞으로 인증 token이 URL로 들어오지 않는다. diff --git a/poc4_active_source_20260714/apps/poc4/mcp_discovery_ui.py b/poc4_active_source_20260714/apps/poc4/mcp_discovery_ui.py index 1ccb827..5fe0060 100644 --- a/poc4_active_source_20260714/apps/poc4/mcp_discovery_ui.py +++ b/poc4_active_source_20260714/apps/poc4/mcp_discovery_ui.py @@ -10,8 +10,6 @@ with the Bearer token entered on the screen. from __future__ import annotations -import base64 -import binascii import hashlib import hmac import html @@ -91,9 +89,8 @@ DEFAULT_VPD_USER_ID = "E1001" VPD_OPERATIONS_URL = "https://hmm-backoffice.cloud-handson.com/" PORTAL_AUTHENTICATED_KEY = "poc4_portal_authenticated" PORTAL_AUTH_USER_KEY = "poc4_portal_auth_user" -PORTAL_LOGIN_FAILURE_KEY = "poc4_portal_login_failed" -PORTAL_REMEMBER_TOKEN_PARAM = "poc4_remember" -PORTAL_REMEMBER_MAX_AGE_SECONDS = 7 * 24 * 60 * 60 +PORTAL_AUTH_PROXY_USER_HEADER = "X-HMM-Authenticated-User" +PORTAL_AUTH_PROXY_EXPIRY_HEADER = "X-HMM-Auth-Expires" AUDIT_DB_ENV_FILE = Path( os.environ.get("POC4_AUDIT_DB_ENV_FILE", str(ENV_FILE)) ).expanduser() @@ -1914,150 +1911,48 @@ def _portal_auth_value(name: str) -> str: return (os.environ.get(name) or _dotenv_value(name)).strip() -def _portal_password_matches(password: str, encoded_password: str) -> bool: +def _proxy_auth_headers() -> tuple[str, int]: try: - scheme, iterations_text, salt_hex, expected_hex = encoded_password.split("$", 3) - iterations = int(iterations_text) - salt = bytes.fromhex(salt_hex) - expected = bytes.fromhex(expected_hex) - except (TypeError, ValueError): - return False - if scheme != "pbkdf2_sha256" or not 100_000 <= iterations <= 2_000_000: - return False - candidate = hashlib.pbkdf2_hmac( - "sha256", - password.encode("utf-8"), - salt, - iterations, - ) - return hmac.compare_digest(candidate, expected) + headers = st.context.headers + username = str(headers.get(PORTAL_AUTH_PROXY_USER_HEADER) or "").strip() + expires_at = int( + str(headers.get(PORTAL_AUTH_PROXY_EXPIRY_HEADER) or "0").strip() + ) + except (AttributeError, TypeError, ValueError): + return "", 0 + return username, expires_at -def _portal_auth_configured() -> bool: - return bool( - _portal_auth_value("POC4_LOGIN_USER") - and _portal_auth_value("POC4_LOGIN_PASSWORD_PBKDF2") - ) - - -def _portal_credentials_are_valid(username: str, password: str) -> bool: +def _restore_portal_proxy_session() -> None: + username, expires_at = _proxy_auth_headers() expected_username = _portal_auth_value("POC4_LOGIN_USER") - encoded_password = _portal_auth_value("POC4_LOGIN_PASSWORD_PBKDF2") - username_matches = hmac.compare_digest(username.strip(), expected_username) - password_matches = _portal_password_matches(password, encoded_password) - return username_matches and password_matches - - -def _portal_remember_secret() -> str: - return _portal_auth_value("POC4_LOGIN_REMEMBER_SECRET") - - -def _portal_remember_token(username: str) -> str: - secret = _portal_remember_secret() - if not secret: - return "" - payload = { - "v": 1, - "u": username.strip(), - "e": int(datetime.now(timezone.utc).timestamp()) + PORTAL_REMEMBER_MAX_AGE_SECONDS, - } - encoded = base64.urlsafe_b64encode( - json.dumps(payload, separators=(",", ":")).encode("utf-8") - ).decode("ascii").rstrip("=") - signature = hmac.new( - secret.encode("utf-8"), encoded.encode("ascii"), hashlib.sha256 - ).hexdigest() - return f"{encoded}.{signature}" - - -def _restore_portal_remembered_session() -> None: - if st.session_state.get(PORTAL_AUTHENTICATED_KEY, False): - return - token = st.query_params.get(PORTAL_REMEMBER_TOKEN_PARAM, "") - if not isinstance(token, str) or not token or len(token) > 2048: - return - secret = _portal_remember_secret() - expected_username = _portal_auth_value("POC4_LOGIN_USER") - try: - encoded, supplied_signature = token.split(".", 1) - expected_signature = hmac.new( - secret.encode("utf-8"), encoded.encode("ascii"), hashlib.sha256 - ).hexdigest() - padded = encoded + "=" * (-len(encoded) % 4) - payload = json.loads(base64.urlsafe_b64decode(padded).decode("utf-8")) - expires_at = int(payload["e"]) - username = str(payload["u"]) - except (binascii.Error, KeyError, TypeError, ValueError, UnicodeDecodeError): - return - if not secret or not hmac.compare_digest(supplied_signature, expected_signature): - return - if expires_at < int(datetime.now(timezone.utc).timestamp()): - return - if not hmac.compare_digest(username, expected_username): + authenticated = bool( + username + and expected_username + and expires_at > int(datetime.now(timezone.utc).timestamp()) + and hmac.compare_digest(username, expected_username) + ) + if not authenticated: + st.session_state.pop(PORTAL_AUTHENTICATED_KEY, None) + st.session_state.pop(PORTAL_AUTH_USER_KEY, None) return st.session_state[PORTAL_AUTHENTICATED_KEY] = True st.session_state[PORTAL_AUTH_USER_KEY] = username - st.session_state[PORTAL_LOGIN_FAILURE_KEY] = False - - -def _clear_portal_remembered_session() -> None: - if PORTAL_REMEMBER_TOKEN_PARAM in st.query_params: - del st.query_params[PORTAL_REMEMBER_TOKEN_PARAM] def _render_portal_login(profile: AppProfile) -> None: with st.container(key="console_login_container"): render_login_brand(st, profile) - if not _portal_auth_configured(): - st.info("데모 계정 설정 중입니다. 운영 담당자에게 계정 발급을 요청해 주세요.") - return - with st.form("poc4_portal_login_form", clear_on_submit=True): - username = st.text_input( - "사용자 ID", - max_chars=80, - placeholder="사용자 ID를 입력하세요.", - ) - password = st.text_input( - "비밀번호", - type="password", - max_chars=200, - placeholder="비밀번호를 입력하세요.", - ) - remember_login = st.checkbox( - "로그인 유지 (7일)", - disabled=not bool(_portal_remember_secret()), - help="이 브라우저에서 7일 동안 로그인 상태를 유지합니다.", - ) - submitted = st.form_submit_button("로그인", use_container_width=True) - if submitted: - if _portal_credentials_are_valid(username, password): - st.session_state[PORTAL_AUTHENTICATED_KEY] = True - st.session_state[PORTAL_AUTH_USER_KEY] = username.strip() - st.session_state[PORTAL_LOGIN_FAILURE_KEY] = False - if remember_login: - token = _portal_remember_token(username) - if token: - st.query_params[PORTAL_REMEMBER_TOKEN_PARAM] = token - else: - _clear_portal_remembered_session() - st.rerun() - st.session_state[PORTAL_LOGIN_FAILURE_KEY] = True - if st.session_state.get(PORTAL_LOGIN_FAILURE_KEY, False): - st.error("사용자 ID 또는 비밀번호를 확인해 주세요.") + st.error( + "인증 게이트웨이의 사용자 확인 정보가 없습니다. " + "공식 포털 주소로 다시 접속해 주세요." + ) st.markdown( f'

{html.escape(profile.login_footer)}

', unsafe_allow_html=True, ) -def _logout_portal() -> None: - _clear_portal_remembered_session() - st.session_state.pop(PORTAL_AUTHENTICATED_KEY, None) - st.session_state.pop(PORTAL_AUTH_USER_KEY, None) - st.session_state.pop(PORTAL_LOGIN_FAILURE_KEY, None) - st.rerun() - - def _render_app_header(profile: AppProfile) -> None: render_console_header(st, profile) @@ -6750,7 +6645,7 @@ def main() -> None: page_title=profile.page_title, page_icon=profile.page_icon, layout="wide" ) _apply_console_theme(profile) - _restore_portal_remembered_session() + _restore_portal_proxy_session() if not st.session_state.get(PORTAL_AUTHENTICATED_KEY, False): _render_portal_login(profile) return @@ -6822,8 +6717,11 @@ def main() -> None: st.caption( f"포털 사용자 · {st.session_state.get(PORTAL_AUTH_USER_KEY, '')}" ) - if st.button("로그아웃", use_container_width=True): - _logout_portal() + st.markdown( + '로그아웃', + unsafe_allow_html=True, + ) st.divider() st.markdown('
AI 사용자 설정
', unsafe_allow_html=True) selected_query_model_profile = st.selectbox( diff --git a/poc4_active_source_20260714/infra/nginx/hmm-portal-auth.locations.conf b/poc4_active_source_20260714/infra/nginx/hmm-portal-auth.locations.conf new file mode 100644 index 0000000..358bc6d --- /dev/null +++ b/poc4_active_source_20260714/infra/nginx/hmm-portal-auth.locations.conf @@ -0,0 +1,41 @@ +location = /auth/check { + internal; + proxy_pass http://127.0.0.1:8621/auth/check; + proxy_pass_request_body off; + proxy_set_header Content-Length ""; + proxy_set_header Cookie $http_cookie; + proxy_set_header X-Real-IP $remote_addr; +} + +location /auth/ { + proxy_pass http://127.0.0.1:8621; + proxy_http_version 1.1; + proxy_set_header Host $host; + proxy_set_header X-Real-IP $remote_addr; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_set_header X-Forwarded-Proto $scheme; +} + +location @hmm_portal_login { + return 302 /auth/login; +} + +location / { + auth_request /auth/check; + error_page 401 = @hmm_portal_login; + auth_request_set $hmm_auth_user $upstream_http_x_auth_user; + auth_request_set $hmm_auth_expires $upstream_http_x_auth_expires; + + proxy_pass http://127.0.0.1:8622; + proxy_http_version 1.1; + proxy_set_header Host $host; + proxy_set_header X-Real-IP $remote_addr; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_set_header X-Forwarded-Proto $scheme; + proxy_set_header X-HMM-Authenticated-User $hmm_auth_user; + proxy_set_header X-HMM-Auth-Expires $hmm_auth_expires; + proxy_set_header Upgrade $http_upgrade; + proxy_set_header Connection "upgrade"; + proxy_read_timeout 300; + proxy_send_timeout 300; +} diff --git a/poc4_active_source_20260714/infra/systemd/hmm-portal-auth.service b/poc4_active_source_20260714/infra/systemd/hmm-portal-auth.service new file mode 100644 index 0000000..80899ee --- /dev/null +++ b/poc4_active_source_20260714/infra/systemd/hmm-portal-auth.service @@ -0,0 +1,29 @@ +[Unit] +Description=HMM Portal HttpOnly Cookie Authentication +After=network-online.target +Wants=network-online.target + +[Service] +Type=simple +User=opc +Group=opc +WorkingDirectory=/opt/hmm-poc4 +Environment=PYTHONUNBUFFERED=1 +EnvironmentFile=/opt/hmm-poc4/.env +ExecStart=/opt/hmm-poc4/.venv/bin/python -m src.agent_console.auth_gateway +Restart=on-failure +RestartSec=3 +NoNewPrivileges=true +PrivateTmp=true +ProtectSystem=strict +ProtectHome=true +ProtectKernelTunables=true +ProtectKernelModules=true +ProtectControlGroups=true +RestrictSUIDSGID=true +LockPersonality=true +RestrictAddressFamilies=AF_INET AF_INET6 +UMask=0077 + +[Install] +WantedBy=multi-user.target diff --git a/poc4_active_source_20260714/src/agent_console/auth_gateway.py b/poc4_active_source_20260714/src/agent_console/auth_gateway.py new file mode 100644 index 0000000..f9be767 --- /dev/null +++ b/poc4_active_source_20260714/src/agent_console/auth_gateway.py @@ -0,0 +1,549 @@ +"""Small localhost authentication service for the HMM Streamlit portal. + +Nginx owns the public security boundary. This module validates the existing +PBKDF2 login, issues a signed HttpOnly cookie, and answers Nginx auth_request +subrequests. Authentication values are never accepted from a URL. +""" + +from __future__ import annotations + +import base64 +import binascii +from collections import defaultdict, deque +from dataclasses import dataclass +from datetime import datetime, timezone +from http import HTTPStatus +from http.cookies import SimpleCookie +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +import hashlib +import hmac +import html +import json +import logging +import os +import secrets +import threading +import time +from typing import Deque +from urllib.parse import parse_qs, urlsplit + + +LOG = logging.getLogger("hmm_portal_auth") +SESSION_COOKIE_NAME = "__Host-HMM_PORTAL_SESSION" +CSRF_COOKIE_NAME = "__Host-HMM_LOGIN_CSRF" +SESSION_TOKEN_VERSION = 2 +MAX_REQUEST_BYTES = 8_192 +MAX_FIELD_CHARS = 200 + + +@dataclass(frozen=True) +class AuthConfig: + username: str + password_pbkdf2: str + cookie_secret: str + bind_address: str = "127.0.0.1" + port: int = 8621 + session_seconds: int = 12 * 60 * 60 + remember_seconds: int = 7 * 24 * 60 * 60 + product_name: str = "HMM AI 업무 에이전트" + login_title: str = "HMM AI 업무 에이전트" + login_description: str = "사용자 인증 후 AI 업무 질의 기능을 이용할 수 있습니다." + login_footer: str = "승인된 사용자만 접속할 수 있습니다." + primary_color: str = "#004b87" + + @classmethod + def from_environment(cls) -> "AuthConfig": + config = cls( + username=os.environ.get("POC4_LOGIN_USER", "").strip(), + password_pbkdf2=os.environ.get( + "POC4_LOGIN_PASSWORD_PBKDF2", "" + ).strip(), + cookie_secret=os.environ.get( + "POC4_LOGIN_COOKIE_SECRET", "" + ).strip(), + bind_address=os.environ.get( + "PORTAL_AUTH_BIND_ADDRESS", "127.0.0.1" + ).strip(), + port=int(os.environ.get("PORTAL_AUTH_PORT", "8621")), + session_seconds=int( + os.environ.get("PORTAL_AUTH_SESSION_SECONDS", str(12 * 60 * 60)) + ), + remember_seconds=int( + os.environ.get( + "PORTAL_AUTH_REMEMBER_SECONDS", str(7 * 24 * 60 * 60) + ) + ), + product_name=os.environ.get( + "AGENT_CONSOLE_NAME", "HMM AI 업무 에이전트" + ).strip(), + login_title=os.environ.get( + "AGENT_CONSOLE_LOGIN_TITLE", "HMM AI 업무 에이전트" + ).strip(), + login_description=os.environ.get( + "AGENT_CONSOLE_LOGIN_DESCRIPTION", + "사용자 인증 후 AI 업무 질의 기능을 이용할 수 있습니다.", + ).strip(), + login_footer=os.environ.get( + "AGENT_CONSOLE_LOGIN_FOOTER", + "승인된 사용자만 접속할 수 있습니다.", + ).strip(), + primary_color=os.environ.get( + "AGENT_CONSOLE_PRIMARY_COLOR", "#004b87" + ).strip(), + ) + config.validate() + return config + + def validate(self) -> None: + if not self.username or not self.password_pbkdf2: + raise ValueError("POC4 portal login credentials are not configured") + if len(self.cookie_secret.encode("utf-8")) < 32: + raise ValueError("POC4_LOGIN_COOKIE_SECRET must be at least 32 bytes") + if self.bind_address not in {"127.0.0.1", "::1"}: + raise ValueError("Portal authentication service must bind to loopback") + if not 1 <= self.port <= 65535: + raise ValueError("PORTAL_AUTH_PORT is invalid") + if not 300 <= self.session_seconds <= 24 * 60 * 60: + raise ValueError("PORTAL_AUTH_SESSION_SECONDS is outside the safe range") + if not self.session_seconds <= self.remember_seconds <= 30 * 24 * 60 * 60: + raise ValueError("PORTAL_AUTH_REMEMBER_SECONDS is outside the safe range") + + +@dataclass(frozen=True) +class AuthenticatedSession: + username: str + expires_at: int + + +class SessionTokenCodec: + def __init__(self, secret: str): + self._secret = secret.encode("utf-8") + + def issue(self, username: str, lifetime_seconds: int, now: int | None = None) -> str: + issued_at = int(time.time()) if now is None else now + payload = { + "v": SESSION_TOKEN_VERSION, + "u": username, + "i": issued_at, + "e": issued_at + lifetime_seconds, + "n": secrets.token_urlsafe(18), + } + encoded = _base64url_encode( + json.dumps(payload, separators=(",", ":")).encode("utf-8") + ) + signature = hmac.new( + self._secret, encoded.encode("ascii"), hashlib.sha256 + ).hexdigest() + return f"{encoded}.{signature}" + + def verify(self, token: str, expected_username: str, now: int | None = None) -> AuthenticatedSession | None: + if not token or len(token) > 2048: + return None + current_time = int(time.time()) if now is None else now + try: + encoded, supplied_signature = token.split(".", 1) + expected_signature = hmac.new( + self._secret, encoded.encode("ascii"), hashlib.sha256 + ).hexdigest() + if not hmac.compare_digest(supplied_signature, expected_signature): + return None + payload = json.loads(_base64url_decode(encoded).decode("utf-8")) + version = int(payload["v"]) + username = str(payload["u"]) + issued_at = int(payload["i"]) + expires_at = int(payload["e"]) + except ( + binascii.Error, + KeyError, + TypeError, + ValueError, + UnicodeDecodeError, + json.JSONDecodeError, + ): + return None + if version != SESSION_TOKEN_VERSION: + return None + if issued_at > current_time + 30 or expires_at <= current_time: + return None + if expires_at - issued_at > 30 * 24 * 60 * 60: + return None + if not hmac.compare_digest(username, expected_username): + return None + return AuthenticatedSession(username=username, expires_at=expires_at) + + +class LoginAttemptLimiter: + def __init__(self, maximum_failures: int = 5, window_seconds: int = 300): + self._maximum_failures = maximum_failures + self._window_seconds = window_seconds + self._failures: dict[str, Deque[float]] = defaultdict(deque) + self._lock = threading.Lock() + + def blocked(self, key: str, now: float | None = None) -> bool: + current_time = time.monotonic() if now is None else now + with self._lock: + failures = self._failures[key] + self._prune(failures, current_time) + return len(failures) >= self._maximum_failures + + def record_failure(self, key: str, now: float | None = None) -> None: + current_time = time.monotonic() if now is None else now + with self._lock: + failures = self._failures[key] + self._prune(failures, current_time) + failures.append(current_time) + + def reset(self, key: str) -> None: + with self._lock: + self._failures.pop(key, None) + + def _prune(self, failures: Deque[float], now: float) -> None: + cutoff = now - self._window_seconds + while failures and failures[0] < cutoff: + failures.popleft() + + +def password_matches(password: str, encoded_password: str) -> bool: + try: + scheme, iterations_text, salt_hex, expected_hex = encoded_password.split( + "$", 3 + ) + iterations = int(iterations_text) + salt = bytes.fromhex(salt_hex) + expected = bytes.fromhex(expected_hex) + except (TypeError, ValueError): + return False + if scheme != "pbkdf2_sha256" or not 100_000 <= iterations <= 2_000_000: + return False + candidate = hashlib.pbkdf2_hmac( + "sha256", password.encode("utf-8"), salt, iterations + ) + return hmac.compare_digest(candidate, expected) + + +def session_cookie_header(token: str, max_age: int | None) -> str: + attributes = [ + f"{SESSION_COOKIE_NAME}={token}", + "Path=/", + "Secure", + "HttpOnly", + "SameSite=Lax", + ] + if max_age is not None: + attributes.append(f"Max-Age={max_age}") + return "; ".join(attributes) + + +def clear_session_cookie_header() -> str: + return ( + f"{SESSION_COOKIE_NAME}=; Path=/; Max-Age=0; " + "Secure; HttpOnly; SameSite=Lax" + ) + + +def csrf_cookie_header(value: str, max_age: int = 600) -> str: + return ( + f"{CSRF_COOKIE_NAME}={value}; Path=/; Max-Age={max_age}; " + "Secure; HttpOnly; SameSite=Strict" + ) + + +def clear_csrf_cookie_header() -> str: + return ( + f"{CSRF_COOKIE_NAME}=; Path=/; Max-Age=0; " + "Secure; HttpOnly; SameSite=Strict" + ) + + +def _base64url_encode(value: bytes) -> str: + return base64.urlsafe_b64encode(value).decode("ascii").rstrip("=") + + +def _base64url_decode(value: str) -> bytes: + padded = value + "=" * (-len(value) % 4) + return base64.urlsafe_b64decode(padded) + + +def _cookie_value(cookie_header: str, name: str) -> str: + try: + cookies = SimpleCookie() + cookies.load(cookie_header) + morsel = cookies.get(name) + return morsel.value if morsel is not None else "" + except (KeyError, TypeError): + return "" + + +def _login_page(config: AuthConfig, csrf_value: str, error: str = "") -> bytes: + error_html = ( + f'' + if error + else "" + ) + return f""" + + + + + {html.escape(config.product_name)} + + + +
+
HMM
+

{html.escape(config.login_title)}

+

{html.escape(config.login_description)}

+ {error_html} +
+ + + + + +
+

{html.escape(config.login_footer)}

+
+ +""".encode("utf-8") + + +def build_handler(config: AuthConfig) -> type[BaseHTTPRequestHandler]: + codec = SessionTokenCodec(config.cookie_secret) + limiter = LoginAttemptLimiter() + + class PortalAuthHandler(BaseHTTPRequestHandler): + server_version = "HMMPortalAuth/1.0" + sys_version = "" + + def do_HEAD(self) -> None: + self._route(send_body=False) + + def do_GET(self) -> None: + self._route(send_body=True) + + def do_POST(self) -> None: + path = urlsplit(self.path).path + if path == "/auth/login": + self._login() + elif path == "/auth/logout": + self._logout() + else: + self._send_text(HTTPStatus.NOT_FOUND, "Not found") + + def _route(self, send_body: bool) -> None: + path = urlsplit(self.path).path + if path == "/auth/check": + self._check() + elif path == "/auth/login": + self._show_login(send_body=send_body) + elif path == "/auth/logout": + self._logout() + elif path == "/auth/healthz": + self._send_text(HTTPStatus.OK, "ok", send_body=send_body) + else: + self._send_text(HTTPStatus.NOT_FOUND, "Not found", send_body=send_body) + + def _check(self) -> None: + session = self._session() + if session is None: + self._send_empty(HTTPStatus.UNAUTHORIZED) + return + self.send_response(HTTPStatus.NO_CONTENT) + self._security_headers() + self.send_header("X-Auth-User", session.username) + self.send_header("X-Auth-Expires", str(session.expires_at)) + self.end_headers() + + def _show_login(self, send_body: bool = True, error: str = "") -> None: + if self._session() is not None and not error: + self._redirect("/") + return + csrf_value = secrets.token_urlsafe(32) + body = _login_page(config, csrf_value, error) + self.send_response(HTTPStatus.OK) + self._security_headers() + self.send_header("Content-Type", "text/html; charset=utf-8") + self.send_header("Content-Length", str(len(body))) + self.send_header("Set-Cookie", csrf_cookie_header(csrf_value)) + self.end_headers() + if send_body: + self.wfile.write(body) + + def _login(self) -> None: + client_key = self._client_key() + if limiter.blocked(client_key): + self._show_login(error="로그인 시도가 잠시 제한되었습니다. 잠시 후 다시 시도해 주세요.") + return + try: + content_length = int(self.headers.get("Content-Length", "0")) + except ValueError: + content_length = 0 + if not 1 <= content_length <= MAX_REQUEST_BYTES: + self._send_text(HTTPStatus.BAD_REQUEST, "Invalid request") + return + raw_body = self.rfile.read(content_length) + try: + form = parse_qs( + raw_body.decode("utf-8"), + keep_blank_values=True, + strict_parsing=False, + max_num_fields=8, + ) + except (UnicodeDecodeError, ValueError): + self._send_text(HTTPStatus.BAD_REQUEST, "Invalid request") + return + username = _form_value(form, "username") + password = _form_value(form, "password") + csrf_form = _form_value(form, "csrf") + csrf_cookie = _cookie_value( + self.headers.get("Cookie", ""), CSRF_COOKIE_NAME + ) + if ( + not csrf_form + or not csrf_cookie + or not hmac.compare_digest(csrf_form, csrf_cookie) + ): + self._send_text(HTTPStatus.BAD_REQUEST, "Invalid request") + return + valid_credentials = ( + len(username) <= 80 + and len(password) <= MAX_FIELD_CHARS + and hmac.compare_digest(username.strip(), config.username) + and password_matches(password, config.password_pbkdf2) + ) + if not valid_credentials: + limiter.record_failure(client_key) + self._show_login(error="사용자 ID 또는 비밀번호를 확인해 주세요.") + return + limiter.reset(client_key) + remember = _form_value(form, "remember") == "yes" + lifetime = ( + config.remember_seconds if remember else config.session_seconds + ) + token = codec.issue(config.username, lifetime) + self.send_response(HTTPStatus.SEE_OTHER) + self._security_headers() + self.send_header("Location", "/") + self.send_header( + "Set-Cookie", + session_cookie_header(token, lifetime if remember else None), + ) + self.send_header("Set-Cookie", clear_csrf_cookie_header()) + self.end_headers() + + def _logout(self) -> None: + self.send_response(HTTPStatus.SEE_OTHER) + self._security_headers() + self.send_header("Location", "/auth/login") + self.send_header("Set-Cookie", clear_session_cookie_header()) + self.send_header("Set-Cookie", clear_csrf_cookie_header()) + self.end_headers() + + def _session(self) -> AuthenticatedSession | None: + token = _cookie_value( + self.headers.get("Cookie", ""), SESSION_COOKIE_NAME + ) + return codec.verify(token, config.username) + + def _client_key(self) -> str: + forwarded = self.headers.get("X-Real-IP", "").strip() + return forwarded or self.client_address[0] + + def _redirect(self, location: str) -> None: + self.send_response(HTTPStatus.SEE_OTHER) + self._security_headers() + self.send_header("Location", location) + self.end_headers() + + def _send_empty(self, status: HTTPStatus) -> None: + self.send_response(status) + self._security_headers() + self.send_header("Content-Length", "0") + self.end_headers() + + def _send_text( + self, + status: HTTPStatus, + message: str, + send_body: bool = True, + ) -> None: + body = message.encode("utf-8") + self.send_response(status) + self._security_headers() + self.send_header("Content-Type", "text/plain; charset=utf-8") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + if send_body: + self.wfile.write(body) + + def _security_headers(self) -> None: + self.send_header("Cache-Control", "no-store") + self.send_header("Pragma", "no-cache") + self.send_header("X-Content-Type-Options", "nosniff") + self.send_header("X-Frame-Options", "DENY") + self.send_header("Referrer-Policy", "no-referrer") + self.send_header( + "Content-Security-Policy", + "default-src 'none'; style-src 'unsafe-inline'; " + "form-action 'self'; base-uri 'none'; frame-ancestors 'none'", + ) + + def log_message(self, _format: str, *args: object) -> None: + # Do not log query strings, cookies, form bodies, or tokens. + LOG.info("%s %s", self.command, urlsplit(self.path).path) + + return PortalAuthHandler + + +def _form_value(form: dict[str, list[str]], name: str) -> str: + values = form.get(name) + return values[0] if values else "" + + +def main() -> None: + logging.basicConfig( + level=os.environ.get("PORTAL_AUTH_LOG_LEVEL", "INFO"), + format="%(asctime)s %(levelname)s %(name)s %(message)s", + ) + config = AuthConfig.from_environment() + server = ThreadingHTTPServer( + (config.bind_address, config.port), build_handler(config) + ) + LOG.info("HMM portal authentication service listening on loopback port %s", config.port) + server.serve_forever() + + +if __name__ == "__main__": + main() diff --git a/poc4_active_source_20260714/src/agent_console/presentation.py b/poc4_active_source_20260714/src/agent_console/presentation.py index 9ed3d23..56fac2e 100644 --- a/poc4_active_source_20260714/src/agent_console/presentation.py +++ b/poc4_active_source_20260714/src/agent_console/presentation.py @@ -94,6 +94,12 @@ def apply_console_theme(st: Any, profile: AppProfile) -> None: .st-key-console_login_container {{ max-width:440px; margin:12vh auto 0; }} .console-login {{ text-align:left; }} .console-login h1 {{ margin:12px 0 8px; font-size:1.7rem; }} + a.console-logout-button {{ display:block; width:100%; padding:.55rem .8rem; + margin:.25rem 0 .75rem; background:#fff; color:var(--console-text) !important; + -webkit-text-fill-color:var(--console-text) !important; + border:1px solid var(--console-border); border-radius:4px; + text-align:center; text-decoration:none; font-weight:700; }} + a.console-logout-button:hover {{ background:#f6f8fa; }} @media (max-width:760px) {{ .block-container {{ padding:1.25rem 1.25rem 3rem; }} .st-key-console_login_container {{ margin-top:8vh; }} }} """, diff --git a/poc4_active_source_20260714/tests/test_auth_gateway.py b/poc4_active_source_20260714/tests/test_auth_gateway.py new file mode 100644 index 0000000..689150a --- /dev/null +++ b/poc4_active_source_20260714/tests/test_auth_gateway.py @@ -0,0 +1,217 @@ +from __future__ import annotations + +import hashlib +import http.client +import os +from pathlib import Path +import sys +import threading +import unittest +from urllib.parse import urlencode + + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT)) + +from src.agent_console.auth_gateway import ( # noqa: E402 + AuthConfig, + LoginAttemptLimiter, + SESSION_COOKIE_NAME, + SessionTokenCodec, + build_handler, + clear_session_cookie_header, + password_matches, + session_cookie_header, +) + + +class AuthGatewayTest(unittest.TestCase): + def setUp(self) -> None: + self.secret = "s" * 48 + self.codec = SessionTokenCodec(self.secret) + + def test_session_token_round_trip_and_tamper_rejection(self) -> None: + token = self.codec.issue("demo-admin", 3600, now=1_000) + + session = self.codec.verify(token, "demo-admin", now=1_001) + + self.assertIsNotNone(session) + self.assertEqual("demo-admin", session.username) + self.assertEqual(4_600, session.expires_at) + self.assertIsNone(self.codec.verify(token + "x", "demo-admin", now=1_001)) + self.assertIsNone(self.codec.verify(token, "other-user", now=1_001)) + + def test_expired_session_token_is_rejected(self) -> None: + token = self.codec.issue("demo-admin", 300, now=1_000) + + self.assertIsNone(self.codec.verify(token, "demo-admin", now=1_300)) + + def test_remember_cookie_has_required_security_attributes(self) -> None: + header = session_cookie_header("signed-value", 604_800) + + self.assertIn(f"{SESSION_COOKIE_NAME}=signed-value", header) + self.assertIn("Path=/", header) + self.assertIn("Secure", header) + self.assertIn("HttpOnly", header) + self.assertIn("SameSite=Lax", header) + self.assertIn("Max-Age=604800", header) + self.assertNotIn("Domain=", header) + + def test_session_cookie_omits_persistent_max_age(self) -> None: + header = session_cookie_header("signed-value", None) + + self.assertNotIn("Max-Age", header) + self.assertIn("HttpOnly", header) + + def test_logout_cookie_expires_immediately(self) -> None: + header = clear_session_cookie_header() + + self.assertIn("Max-Age=0", header) + self.assertIn("Secure", header) + self.assertIn("HttpOnly", header) + + def test_pbkdf2_password_verification(self) -> None: + salt = bytes.fromhex("00112233445566778899aabbccddeeff") + expected = hashlib.pbkdf2_hmac( + "sha256", b"correct-password", salt, 200_000 + ).hex() + encoded = f"pbkdf2_sha256$200000${salt.hex()}${expected}" + + self.assertTrue(password_matches("correct-password", encoded)) + self.assertFalse(password_matches("wrong-password", encoded)) + + def test_rate_limiter_blocks_only_after_threshold(self) -> None: + limiter = LoginAttemptLimiter(maximum_failures=2, window_seconds=10) + + limiter.record_failure("client", now=1) + self.assertFalse(limiter.blocked("client", now=2)) + limiter.record_failure("client", now=3) + self.assertTrue(limiter.blocked("client", now=4)) + self.assertFalse(limiter.blocked("client", now=20)) + + def test_environment_config_requires_new_cookie_secret(self) -> None: + previous = dict(os.environ) + try: + os.environ["POC4_LOGIN_USER"] = "demo-admin" + os.environ["POC4_LOGIN_PASSWORD_PBKDF2"] = "encoded" + os.environ.pop("POC4_LOGIN_COOKIE_SECRET", None) + with self.assertRaisesRegex(ValueError, "COOKIE_SECRET"): + AuthConfig.from_environment() + finally: + os.environ.clear() + os.environ.update(previous) + + def test_http_login_check_and_logout_flow_never_uses_url_token(self) -> None: + salt = bytes.fromhex("00112233445566778899aabbccddeeff") + expected = hashlib.pbkdf2_hmac( + "sha256", b"correct-password", salt, 200_000 + ).hex() + config = AuthConfig( + username="demo-admin", + password_pbkdf2=( + f"pbkdf2_sha256$200000${salt.hex()}${expected}" + ), + cookie_secret=self.secret, + port=8621, + ) + from http.server import ThreadingHTTPServer + + server = ThreadingHTTPServer(("127.0.0.1", 0), build_handler(config)) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + connection = http.client.HTTPConnection( + "127.0.0.1", server.server_address[1], timeout=3 + ) + try: + connection.request("GET", "/auth/login") + login_page = connection.getresponse() + body = login_page.read().decode("utf-8") + self.assertEqual(200, login_page.status) + csrf_header = next( + value + for name, value in login_page.getheaders() + if name.lower() == "set-cookie" + and value.startswith("__Host-HMM_LOGIN_CSRF=") + ) + csrf_value = csrf_header.split("=", 1)[1].split(";", 1)[0] + self.assertIn( + f'name="csrf" value="{csrf_value}"', + body, + ) + + payload = urlencode( + { + "csrf": csrf_value, + "username": "demo-admin", + "password": "correct-password", + "remember": "yes", + } + ) + connection.request( + "POST", + "/auth/login", + body=payload, + headers={ + "Content-Type": "application/x-www-form-urlencoded", + "Cookie": f"__Host-HMM_LOGIN_CSRF={csrf_value}", + }, + ) + logged_in = connection.getresponse() + logged_in.read() + self.assertEqual(303, logged_in.status) + self.assertEqual("/", logged_in.getheader("Location")) + self.assertNotRegex(logged_in.getheader("Location"), r"token|remember") + session_header = next( + value + for name, value in logged_in.getheaders() + if name.lower() == "set-cookie" + and value.startswith(f"{SESSION_COOKIE_NAME}=") + ) + session_value = session_header.split("=", 1)[1].split(";", 1)[0] + self.assertIn("Secure", session_header) + self.assertIn("HttpOnly", session_header) + self.assertIn("SameSite=Lax", session_header) + + connection.request( + "GET", + "/auth/check", + headers={"Cookie": f"{SESSION_COOKIE_NAME}={session_value}"}, + ) + check = connection.getresponse() + check.read() + self.assertEqual(204, check.status) + self.assertEqual("demo-admin", check.getheader("X-Auth-User")) + + connection.request( + "GET", + "/auth/check", + headers={"Cookie": f"{SESSION_COOKIE_NAME}={session_value}x"}, + ) + tampered = connection.getresponse() + tampered.read() + self.assertEqual(401, tampered.status) + + connection.request( + "GET", + "/auth/logout", + headers={"Cookie": f"{SESSION_COOKIE_NAME}={session_value}"}, + ) + logout = connection.getresponse() + logout.read() + self.assertEqual(303, logout.status) + self.assertEqual("/auth/login", logout.getheader("Location")) + self.assertTrue( + any( + name.lower() == "set-cookie" and "Max-Age=0" in value + for name, value in logout.getheaders() + ) + ) + finally: + connection.close() + server.shutdown() + server.server_close() + thread.join(timeout=3) + + +if __name__ == "__main__": + unittest.main()