From 6c87660451a89442ad1201d58032e66b7ffdac8c Mon Sep 17 00:00:00 2001 From: phil Date: Tue, 6 Oct 2026 10:50:26 +0900 Subject: [PATCH] fix(security): Hash passwords with scrypt instead of SHA-256 CodeQL py/weak-sensitive-data-hashing: gateway login, user creation and tenant admin creation stored unsalted SHA-256 of the password, which is cheap to brute-force offline. Add password_hasher (stdlib scrypt, random salt, constant-time compare) and use it in all three places. Existing SHA-256 hashes still verify and are replaced with scrypt on the next successful login, so no user has to reset a password. The E2E seed now uses the same hasher instead of its own SHA-256 copy. Co-Authored-By: Claude Fable 5.1 Signed-off-by: phil --- .../oneerp_gateway_app/routes/admin.py | 4 +- .../services/auth_service.py | 15 ++-- .../services/password_hasher.py | 71 +++++++++++++++++++ .../services/user_service.py | 8 +-- .../gateway/tests/unit/test_auth_service.py | 33 +++++++++ .../tests/unit/test_password_hasher.py | 38 ++++++++++ .../platform/gateway/tests/unit/test_users.py | 7 +- tests/e2e/helpers/seed.py | 9 +-- 8 files changed, 158 insertions(+), 27 deletions(-) create mode 100644 services/platform/gateway/oneerp_gateway_app/services/password_hasher.py create mode 100644 services/platform/gateway/tests/unit/test_password_hasher.py diff --git a/services/platform/gateway/oneerp_gateway_app/routes/admin.py b/services/platform/gateway/oneerp_gateway_app/routes/admin.py index 8cd303d..0bd2ad8 100644 --- a/services/platform/gateway/oneerp_gateway_app/routes/admin.py +++ b/services/platform/gateway/oneerp_gateway_app/routes/admin.py @@ -2,7 +2,6 @@ from __future__ import annotations -import hashlib from datetime import UTC, datetime from typing import Any @@ -17,6 +16,7 @@ from ..audit_hooks import emit as audit_emit from ..models.tenant import Tenant, TenantCreate, TenantUpdate from ..models.user import User, UserTier +from ..services.password_hasher import hash_password router = APIRouter( prefix="/api/v1/admin", @@ -176,7 +176,7 @@ def create_admin_user(tenant_id: str, body: AdminUserCreateRequest) -> dict[str, user_repo = _get_user_repo(tenant_id) doc_id = generate_name(_USER_PREFIX) - password_hash = hashlib.sha256(body.password.encode()).hexdigest() if body.password else "" + password_hash = hash_password(body.password) if body.password else "" user = User( _id=doc_id, diff --git a/services/platform/gateway/oneerp_gateway_app/services/auth_service.py b/services/platform/gateway/oneerp_gateway_app/services/auth_service.py index 65676b4..725c181 100644 --- a/services/platform/gateway/oneerp_gateway_app/services/auth_service.py +++ b/services/platform/gateway/oneerp_gateway_app/services/auth_service.py @@ -2,7 +2,6 @@ from __future__ import annotations -import hashlib from datetime import UTC, datetime from typing import TYPE_CHECKING, Any @@ -11,6 +10,7 @@ from oneerp_core.errors import raise_forbidden, raise_unauthorized from oneerp_core.repository import Repository +from .password_hasher import hash_password, needs_rehash, verify_password from .token_service import ( clear_auth_cookies, create_access_token, @@ -31,11 +31,6 @@ def _require_tenant_header(request: Request) -> str: return tenant_id -def _hash_password(password: str) -> str: - """비밀번호를 SHA-256으로 해시한다.""" - return hashlib.sha256(password.encode()).hexdigest() - - def login(*, username: str, password: str, request: Request, response: Response) -> dict[str, Any]: """로그인 유스케이스를 수행한다.""" tenant_id = _require_tenant_header(request) @@ -52,7 +47,7 @@ def login(*, username: str, password: str, request: Request, response: Response) stored_hash = user_doc.get("password_hash", "") if not stored_hash: raise_unauthorized("비밀번호가 아직 설정되지 않았습니다") - if stored_hash != _hash_password(password): + if not verify_password(password, stored_hash): raise_unauthorized("사용자명 또는 비밀번호가 올바르지 않습니다") if not user_doc.get("is_active", True): @@ -67,8 +62,12 @@ def login(*, username: str, password: str, request: Request, response: Response) expires_in=expires_in, ) + # 레거시 해시는 로그인 성공 시 scrypt 로 교체한다. + updates: dict[str, Any] = {"last_login": datetime.now(tz=UTC)} + if needs_rehash(stored_hash): + updates["password_hash"] = hash_password(password) if user_doc.get("_id"): - repo.update_by_id(str(user_doc["_id"]), {"last_login": datetime.now(tz=UTC)}) + repo.update_by_id(str(user_doc["_id"]), updates) return {"access_token": access_token, "expires_in": expires_in} diff --git a/services/platform/gateway/oneerp_gateway_app/services/password_hasher.py b/services/platform/gateway/oneerp_gateway_app/services/password_hasher.py new file mode 100644 index 0000000..b16a731 --- /dev/null +++ b/services/platform/gateway/oneerp_gateway_app/services/password_hasher.py @@ -0,0 +1,71 @@ +"""비밀번호 해시 — scrypt + 무작위 salt. + +저장 형식: ``scrypt$$$

$$`` +예: ``scrypt$16384$8$1$9f0c…$4be1…`` + +레거시 SHA-256 hex 해시는 검증만 지원하고, 로그인 성공 시 scrypt 로 교체한다 +(``needs_rehash``). 새 해시는 언제나 scrypt 다. +""" + +from __future__ import annotations + +import hashlib +import hmac +import secrets + +_SCHEME = "scrypt" +_SEP = "$" +_SCRYPT_N = 2**14 +_SCRYPT_R = 8 +_SCRYPT_P = 1 +_SALT_BYTES = 16 +_KEY_BYTES = 32 +_FIELD_COUNT = 6 +_LEGACY_HEX_LEN = 64 + + +def _derive(password: str, salt: bytes, n: int, r: int, p: int) -> bytes: + return hashlib.scrypt(password.encode(), salt=salt, n=n, r=r, p=p, dklen=_KEY_BYTES) + + +def hash_password(password: str) -> str: + """비밀번호를 scrypt 로 해시한다.""" + salt = secrets.token_bytes(_SALT_BYTES) + key = _derive(password, salt, _SCRYPT_N, _SCRYPT_R, _SCRYPT_P) + fields = [_SCHEME, str(_SCRYPT_N), str(_SCRYPT_R), str(_SCRYPT_P), salt.hex(), key.hex()] + return _SEP.join(fields) + + +def _verify_scrypt(password: str, stored: str) -> bool: + fields = stored.split(_SEP) + if len(fields) != _FIELD_COUNT: + return False + + _, n, r, p, salt_hex, key_hex = fields + try: + key = _derive(password, bytes.fromhex(salt_hex), int(n), int(r), int(p)) + expected = bytes.fromhex(key_hex) + except ValueError: + return False + return hmac.compare_digest(key, expected) + + +def _verify_legacy(password: str, stored: str) -> bool: + # 마이그레이션 전용 — 기존 SHA-256 저장분을 확인해 scrypt 로 올리기 위해서만 쓴다. + digest = hashlib.sha256(password.encode()).hexdigest() + return hmac.compare_digest(digest, stored) + + +def verify_password(password: str, stored: str) -> bool: + """저장된 해시(scrypt 또는 레거시 SHA-256)와 비밀번호를 상수 시간 비교한다.""" + if stored.startswith(_SCHEME + _SEP): + return _verify_scrypt(password, stored) + if len(stored) == _LEGACY_HEX_LEN: + return _verify_legacy(password, stored) + return False + + +def needs_rehash(stored: str) -> bool: + """현재 scrypt 파라미터가 아닌 해시면 True — 로그인 성공 시 교체 대상.""" + prefix = _SEP.join([_SCHEME, str(_SCRYPT_N), str(_SCRYPT_R), str(_SCRYPT_P)]) + _SEP + return not stored.startswith(prefix) diff --git a/services/platform/gateway/oneerp_gateway_app/services/user_service.py b/services/platform/gateway/oneerp_gateway_app/services/user_service.py index 5d2b6f5..0914109 100644 --- a/services/platform/gateway/oneerp_gateway_app/services/user_service.py +++ b/services/platform/gateway/oneerp_gateway_app/services/user_service.py @@ -2,7 +2,6 @@ from __future__ import annotations -import hashlib from datetime import UTC, datetime from typing import TYPE_CHECKING, Any @@ -11,6 +10,7 @@ from oneerp_core.repository import Repository from oneerp_gateway_app.models.user import User, UserAuthProvider, UserInvitationStatus +from oneerp_gateway_app.services.password_hasher import hash_password from oneerp_gateway_app.services.user_presenter import ( build_summary, decorate_user, @@ -33,10 +33,6 @@ def _get_company_repo(tenant_id: str) -> Repository: return Repository(_COMPANY_COLLECTION, tenant_id=tenant_id) -def _hash_password(password: str) -> str: - return hashlib.sha256(password.encode()).hexdigest() - - def normalize_auth_provider(provider: Any, oidc_subject: str) -> str: if provider: return str(provider) @@ -141,7 +137,7 @@ def prepare_user_payload( ) if password: - payload["password_hash"] = _hash_password(password) + payload["password_hash"] = hash_password(password) return payload diff --git a/services/platform/gateway/tests/unit/test_auth_service.py b/services/platform/gateway/tests/unit/test_auth_service.py index 46ac802..1b23334 100644 --- a/services/platform/gateway/tests/unit/test_auth_service.py +++ b/services/platform/gateway/tests/unit/test_auth_service.py @@ -10,6 +10,7 @@ from oneerp_core.config import get_core_settings from oneerp_core.errors import OneERPError from oneerp_gateway_app.services.auth_service import login, refresh +from oneerp_gateway_app.services.password_hasher import verify_password from oneerp_gateway_app.services.token_service import create_access_token @@ -112,3 +113,35 @@ def test_refresh는_type이_refresh가_아니면_거부한다(mock_decode_token: assert exc_info.value.status_code == 401 assert "유효하지 않은 리프레시 토큰" in str(exc_info.value.detail) + + +@patch("oneerp_gateway_app.services.permission_service.Repository") +@patch("oneerp_gateway_app.services.auth_service.Repository") +def test_login은_레거시_해시를_scrypt로_교체한다( + mock_repo_cls: MagicMock, + mock_permission_repo_cls: MagicMock, +) -> None: + user_repo = MagicMock() + user_repo.find_many.return_value = [ + { + "_id": "USR-001", + "username": "demo", + "tenant_id": "default", + "roles": ["admin"], + "password_hash": "2bb80d537b1da3e38bd30361aa855686bde0eacd7162fef6a25fe97bf527a25b", + "is_active": True, + } + ] + mock_repo_cls.return_value = user_repo + mock_permission_repo_cls.return_value.find_many.return_value = [] + + login( + username="demo", + password="secret", # noqa: S106 + request=_request_with_headers(), + response=Response(), + ) + + updates = user_repo.update_by_id.call_args.args[1] + assert updates["password_hash"].startswith("scrypt$") + assert verify_password("secret", updates["password_hash"]) diff --git a/services/platform/gateway/tests/unit/test_password_hasher.py b/services/platform/gateway/tests/unit/test_password_hasher.py new file mode 100644 index 0000000..296f278 --- /dev/null +++ b/services/platform/gateway/tests/unit/test_password_hasher.py @@ -0,0 +1,38 @@ +"""password_hasher 단위 테스트.""" + +from __future__ import annotations + +from oneerp_gateway_app.services.password_hasher import ( + hash_password, + needs_rehash, + verify_password, +) + +_PLAIN = "secret" +# 레거시 SHA-256("secret") — 마이그레이션 전 저장분 형식. +_LEGACY = "2bb80d537b1da3e38bd30361aa855686bde0eacd7162fef6a25fe97bf527a25b" + + +def test_scrypt_해시는_검증되고_재해시가_불필요하다() -> None: + stored = hash_password(_PLAIN) + + assert stored.startswith("scrypt$") + assert verify_password(_PLAIN, stored) + assert not verify_password("wrong", stored) + assert not needs_rehash(stored) + + +def test_같은_비밀번호도_salt로_해시가_다르다() -> None: + assert hash_password(_PLAIN) != hash_password(_PLAIN) + + +def test_레거시_해시는_검증되고_재해시_대상이다() -> None: + assert verify_password(_PLAIN, _LEGACY) + assert not verify_password("wrong", _LEGACY) + assert needs_rehash(_LEGACY) + + +def test_형식이_깨진_해시는_거부한다() -> None: + assert not verify_password(_PLAIN, "") + assert not verify_password(_PLAIN, "scrypt$bad") + assert not verify_password(_PLAIN, "scrypt$16384$8$1$zz$zz") diff --git a/services/platform/gateway/tests/unit/test_users.py b/services/platform/gateway/tests/unit/test_users.py index b6144e2..c473e0b 100644 --- a/services/platform/gateway/tests/unit/test_users.py +++ b/services/platform/gateway/tests/unit/test_users.py @@ -8,6 +8,7 @@ from fastapi.testclient import TestClient from oneerp_core.errors import OneERPError, oneerp_error_handler from oneerp_gateway_app.routes.users import router +from oneerp_gateway_app.services.password_hasher import needs_rehash, verify_password _app = FastAPI() _app.add_exception_handler(OneERPError, oneerp_error_handler) # type: ignore[arg-type] @@ -59,10 +60,8 @@ def test_사용자_생성은_초대상태와_해시를_저장한다( assert response.json()["id"] == "USR-2026-00001" mock_name.assert_called_once_with("USR", tenant_id="test-tenant") inserted_doc = mock_repo.return_value.insert.call_args.args[0] - assert ( - inserted_doc.password_hash - == "94e0f9bc7f5a5225bd141bad5adf9befcc112aef09b88f47a14e20b75a7bbec2" # noqa: S105 - ) + assert verify_password("Secret123!", inserted_doc.password_hash) + assert not needs_rehash(inserted_doc.password_hash) assert inserted_doc.invitation_status == "pending" assert inserted_doc.auth_provider == "oidc" assert inserted_doc.company_id == "COMP-001" diff --git a/tests/e2e/helpers/seed.py b/tests/e2e/helpers/seed.py index cb487a3..7760395 100644 --- a/tests/e2e/helpers/seed.py +++ b/tests/e2e/helpers/seed.py @@ -6,10 +6,10 @@ from __future__ import annotations -import hashlib from datetime import UTC, datetime from typing import Any +from oneerp_gateway_app.services.password_hasher import hash_password from pymongo import MongoClient E2E_TENANT_ID = "default" @@ -17,11 +17,6 @@ E2E_PASSWORD = "e2e-pass-2026" # noqa: S105 -def _hash_password(password: str) -> str: - """비밀번호를 SHA-256으로 해시한다 (gateway auth.py와 동일).""" - return hashlib.sha256(password.encode()).hexdigest() - - def seed_minimal_auth(client: MongoClient, db_name: str) -> dict[str, Any]: """로그인 가능한 최소 상태를 시드한다. @@ -87,7 +82,7 @@ def seed_minimal_auth(client: MongoClient, db_name: str) -> dict[str, Any]: "is_super_admin": True, "is_active": True, "auth_provider": "password", - "password_hash": _hash_password(E2E_PASSWORD), + "password_hash": hash_password(E2E_PASSWORD), "created_at": now, }, },