366 lines
16 KiB
Python
366 lines
16 KiB
Python
import uuid
|
|
|
|
from fastapi import Depends
|
|
|
|
from common.database.db_session_manager import DB_SESSION_MNG
|
|
from common.database.model.models import supplier_user_tokens, supplier_users, suppliers
|
|
from common.enums import AccountStatus, DBWRType, ErrorType, TokenType
|
|
from common.logger import LOG
|
|
from common.models.gmodel import UserInfo
|
|
from common.utils.gtime import GTime
|
|
from config.server_configs import jwt_token_config
|
|
from crud.user_crud import IUserCRUD, UserCRUD
|
|
from router.v1.auth.protocol import (
|
|
Res_CreateAccount,
|
|
Res_HidePopup,
|
|
Res_Login,
|
|
Res_Logout,
|
|
Res_Me,
|
|
Res_PopupStatus,
|
|
Res_RefreshToken,
|
|
Res_SessionBranding,
|
|
)
|
|
from router.v1.validator.dependencies import CreateAccessToken, CreateRefreshToken, GetHashedPW, VerifyPW
|
|
|
|
|
|
class AuthService:
|
|
"""비즈니스 로직 계층 (MVC 의 컨트롤러-서비스 분리에서 서비스).
|
|
|
|
- 유저는 supplier_users 테이블(JWT subject = UserInfo). uuid 는 문자열로 인코딩.
|
|
- CRUD 는 Depends 로 인터페이스 타입으로 주입받는다.
|
|
- DB 접근은 DB_SESSION_MNG 의 람다 실행으로만 한다.
|
|
조회 = execute_lambda(..., DB_READ, lambda s: crud.xxx(s, ...))
|
|
변경 = execute_lambda_run([DBType], [lambda s: crud.xxx(s, ...)])
|
|
- 모든 메서드는 Res_* 를 만들어 result 에 ErrorType 을 세팅해 반환한다.
|
|
"""
|
|
|
|
def __init__(self, user_crud: IUserCRUD = Depends(UserCRUD)):
|
|
self.user_crud = user_crud
|
|
|
|
async def create_account(
|
|
self, supplier_id: str, id: str, pw: str, name: str, email: str, contact_number: str, role: int, connect_ip: str
|
|
) -> Res_CreateAccount:
|
|
LOG.i(f"CREATE : {id=}, {supplier_id=}")
|
|
res = Res_CreateAccount()
|
|
|
|
# 0) supplier_id 형식 검증
|
|
try:
|
|
sid = uuid.UUID(supplier_id)
|
|
except (ValueError, TypeError):
|
|
res.result.SetResult(ErrorType.INVALID_REQUEST_DATA)
|
|
return res
|
|
|
|
# 1) 공급사 존재 확인 (no-FK 라 앱에서 무결성 검증, PARTNER Read)
|
|
err_type, _ = await DB_SESSION_MNG.execute_lambda(
|
|
suppliers.DBType(),
|
|
DBWRType.DB_READ.value,
|
|
lambda s: self.user_crud.get_supplier_name(s, sid),
|
|
)
|
|
if err_type != ErrorType.SUCCESS:
|
|
# 존재하지 않는 supplier_id
|
|
res.result.SetResult(ErrorType.INVALID_REQUEST_DATA)
|
|
return res
|
|
|
|
# 2) 중복 로그인 ID 확인 (USER Read)
|
|
err_type = await DB_SESSION_MNG.execute_lambda(
|
|
supplier_users.DBType(),
|
|
DBWRType.DB_READ.value,
|
|
lambda s: self.user_crud.is_account(s, id),
|
|
)
|
|
if err_type == ErrorType.DB_ALREADY_SAME_KEY:
|
|
res.result.SetResult(ErrorType.ACCOUNT_ALREADY_EXIST)
|
|
return res
|
|
if err_type != ErrorType.SUCCESS:
|
|
res.result.SetResult(err_type)
|
|
return res
|
|
|
|
# 3) 생성 (pw bcrypt 해시. last_accessed_at 은 NOT NULL/무기본값이라 생성 시각으로 둔다)
|
|
account = supplier_users(
|
|
supplier_id=sid,
|
|
id=id,
|
|
password=await GetHashedPW(pw),
|
|
name=name or None,
|
|
email=email or None,
|
|
contact_number=contact_number or None,
|
|
last_accessed_at=GTime.UTC(),
|
|
status=AccountStatus.ACTIVE.value,
|
|
role=role,
|
|
)
|
|
err_type = await DB_SESSION_MNG.execute_lambda_run(
|
|
[supplier_users.DBType()],
|
|
[lambda s: self.user_crud.add_account(s, account)],
|
|
)
|
|
if err_type != ErrorType.SUCCESS:
|
|
# 사전 검사와 INSERT 사이 경쟁 조건의 unique 위반은 동일 코드로 매핑.
|
|
if err_type == ErrorType.DB_ALREADY_SAME_KEY:
|
|
res.result.SetResult(ErrorType.ACCOUNT_ALREADY_EXIST)
|
|
else:
|
|
res.result.SetResult(err_type)
|
|
return res
|
|
|
|
res.su_id = str(account.su_id)
|
|
return res
|
|
|
|
async def attempt_login(self, id: str, pw: str, connect_ip: str) -> Res_Login:
|
|
LOG.i(f"LOGIN : {id=}")
|
|
res = Res_Login()
|
|
|
|
# 1) 계정 조회 (USER Read 세션)
|
|
err_type, account = await DB_SESSION_MNG.execute_lambda(
|
|
supplier_users.DBType(),
|
|
DBWRType.DB_READ.value,
|
|
lambda s: self.user_crud.get_account_by_id(s, id),
|
|
)
|
|
if err_type != ErrorType.SUCCESS:
|
|
# 계정 없음/조회 실패 모두 로그인 실패로 일반화
|
|
res.result.SetResult(ErrorType.ACCOUNT_INVALID_INFO)
|
|
return res
|
|
account: supplier_users
|
|
|
|
# 2) 비밀번호 검증
|
|
if not await VerifyPW(pw, account.password):
|
|
res.result.SetResult(ErrorType.ACCOUNT_INVALID_INFO)
|
|
return res
|
|
|
|
# 3) 상태 확인 (active 만 허용)
|
|
if account.status != AccountStatus.ACTIVE.value:
|
|
res.result.SetResult(ErrorType.ACCOUNT_BLOCKED_USER)
|
|
return res
|
|
|
|
# 3-1) 공급사명 조회 (PARTNER Read 세션). 부가 정보라 실패해도 로그인은 막지 않고 빈 값.
|
|
_, sname = await DB_SESSION_MNG.execute_lambda(
|
|
suppliers.DBType(),
|
|
DBWRType.DB_READ.value,
|
|
lambda s: self.user_crud.get_supplier_name(s, account.supplier_id),
|
|
)
|
|
supplier_name = sname or ""
|
|
|
|
# 4) 토큰 발급
|
|
user_info = UserInfo(
|
|
su_id=str(account.su_id),
|
|
id=account.id,
|
|
name=account.name or "",
|
|
supplier_id=str(account.supplier_id),
|
|
supplier_name=supplier_name,
|
|
role=account.role,
|
|
)
|
|
res.access_token = CreateAccessToken(user_info)
|
|
res.refresh_token = CreateRefreshToken(user_info)
|
|
|
|
# 5) 마지막 접속 시간 갱신 + 토큰 교체 (Write DB, 한 트랜잭션)
|
|
# 단일 세션: 이전 토큰 행을 모두 지우고 access/refresh 2행을 새로 넣어 이전 세션을 무효화한다.
|
|
# 저장된 토큰은 추후 로그아웃/검증(토큰 대조)에서 사용한다.
|
|
now = GTime.UTC()
|
|
access_row = supplier_user_tokens(
|
|
su_id=account.su_id,
|
|
type=TokenType.ACCESS.value,
|
|
token={"jwt": res.access_token},
|
|
issued_at=now,
|
|
expired_at=GTime.AddMinutes(jwt_token_config.access_expire_min),
|
|
)
|
|
refresh_row = supplier_user_tokens(
|
|
su_id=account.su_id,
|
|
type=TokenType.REFRESH.value,
|
|
token={"jwt": res.refresh_token},
|
|
issued_at=now,
|
|
expired_at=GTime.AddDays(jwt_token_config.refresh_expire_day),
|
|
)
|
|
err_type = await DB_SESSION_MNG.execute_lambda_run(
|
|
[supplier_users.DBType()],
|
|
[
|
|
lambda s: self.user_crud.update_last_accessed(s, account.su_id),
|
|
lambda s: self.user_crud.delete_tokens_by_su_id(s, account.su_id), # 단일 세션: 이전 토큰 제거
|
|
lambda s: self.user_crud.add_token(s, access_row),
|
|
lambda s: self.user_crud.add_token(s, refresh_row),
|
|
],
|
|
)
|
|
if err_type != ErrorType.SUCCESS:
|
|
res.result.SetResult(err_type)
|
|
return res
|
|
|
|
res.su_id = str(account.su_id)
|
|
res.name = account.name or ""
|
|
res.supplier_id = str(account.supplier_id)
|
|
res.supplier_name = supplier_name
|
|
res.role = account.role
|
|
return res
|
|
|
|
async def __load_active_account(self, su_id_str: str) -> tuple[ErrorType, UserInfo]:
|
|
"""su_id 로 유저를 조회해 존재 + status=active 확인 후, 공급사명까지 채운 DB 최신값 UserInfo 를
|
|
반환한다. (토큰 발급 후 삭제/비활성된 계정 차단용)
|
|
실패 시 (에러코드, None) 을 반환하며, HTTP 변환은 라우터가 result 로 내려보낸다.
|
|
"""
|
|
# 1) 계정 조회 (USER Read 세션)
|
|
err_type, account = await DB_SESSION_MNG.execute_lambda(
|
|
supplier_users.DBType(),
|
|
DBWRType.DB_READ.value,
|
|
lambda s: self.user_crud.get_account_by_su_id(s, uuid.UUID(su_id_str)),
|
|
)
|
|
if err_type != ErrorType.SUCCESS or account is None:
|
|
return ErrorType.ACCOUNT_INVALID_INFO, None
|
|
if account.status != AccountStatus.ACTIVE.value: # active 만 허용
|
|
return ErrorType.ACCOUNT_BLOCKED_USER, None
|
|
|
|
# 2) 공급사명 조회 (PARTNER Read 세션). 부가 정보라 실패해도 빈 값.
|
|
_, sname = await DB_SESSION_MNG.execute_lambda(
|
|
suppliers.DBType(),
|
|
DBWRType.DB_READ.value,
|
|
lambda s: self.user_crud.get_supplier_name(s, account.supplier_id),
|
|
)
|
|
return ErrorType.SUCCESS, UserInfo(
|
|
su_id=str(account.su_id),
|
|
id=account.id,
|
|
name=account.name or "",
|
|
supplier_id=str(account.supplier_id),
|
|
supplier_name=sname or "",
|
|
role=account.role,
|
|
)
|
|
|
|
async def __verify_stored_token(self, su_id_str: str, token_type: int, presented: str) -> bool:
|
|
"""제시된 토큰이 저장된 토큰과 일치하는지 확인한다(stateful 단일 세션).
|
|
로그아웃·타기기 로그인으로 교체되면 저장 토큰이 없거나 달라져 False 가 된다.
|
|
"""
|
|
err_type, stored = await DB_SESSION_MNG.execute_lambda(
|
|
supplier_users.DBType(),
|
|
DBWRType.DB_READ.value,
|
|
lambda s: self.user_crud.get_token(s, uuid.UUID(su_id_str), token_type),
|
|
)
|
|
return err_type == ErrorType.SUCCESS and stored == presented
|
|
|
|
async def authenticate(self, user_info: UserInfo, access_token: str) -> tuple[ErrorType, UserInfo]:
|
|
"""access 토큰 보호 요청 공통 인증: 계정 활성 확인 + 저장된 access 토큰 대조.
|
|
성공 시 (SUCCESS, DB 최신 UserInfo), 실패 시 (에러코드, None). 다른 도메인 service 에서도 재사용한다.
|
|
"""
|
|
err_type, info = await self.__load_active_account(user_info.su_id)
|
|
if err_type != ErrorType.SUCCESS:
|
|
return err_type, None
|
|
if not await self.__verify_stored_token(info.su_id, TokenType.ACCESS.value, access_token):
|
|
return ErrorType.TOKEN_REVOKED, None # 로그아웃/타기기 로그인으로 무효화됨
|
|
return ErrorType.SUCCESS, info
|
|
|
|
async def get_me(self, user_info: UserInfo, access_token: str) -> Res_Me:
|
|
# 토큰 디코드는 라우터 Depends(IsValidAccessToken) 에서 수행됨. 여기선 공통 인증으로 검증.
|
|
res = Res_Me()
|
|
err_type, info = await self.authenticate(user_info, access_token)
|
|
if err_type != ErrorType.SUCCESS:
|
|
res.result.SetResult(err_type)
|
|
return res
|
|
res.su_id = info.su_id
|
|
res.id = info.id
|
|
res.name = info.name
|
|
res.supplier_id = info.supplier_id
|
|
res.supplier_name = info.supplier_name
|
|
res.role = info.role
|
|
# 소속 회사 설정(companies.settings) — 로고/서비스명(branding) + 협상완료 부가필드(session_fields). 실패해도 기본값.
|
|
_e, settings = await DB_SESSION_MNG.execute_lambda(
|
|
suppliers.DBType(),
|
|
DBWRType.DB_READ.value,
|
|
lambda s: self.user_crud.get_company_settings(s, uuid.UUID(info.supplier_id)),
|
|
)
|
|
settings = settings or {}
|
|
res.branding = settings.get("branding") or {}
|
|
res.session_fields = settings.get("session_fields") or []
|
|
res.guide_notices = settings.get("guide_notices") or []
|
|
return res
|
|
|
|
async def session_branding(self, session_id: str) -> Res_SessionBranding:
|
|
"""로그인 전(초청 링크 진입) 화면용 브랜딩. 인증 없이 session_id 로만 조회하며 브랜딩 외 정보는 내리지 않는다."""
|
|
res = Res_SessionBranding()
|
|
try:
|
|
sid = uuid.UUID(session_id)
|
|
except ValueError:
|
|
res.result.SetResult(ErrorType.INVALID_REQUEST_DATA)
|
|
return res
|
|
_e, branding = await DB_SESSION_MNG.execute_lambda(
|
|
suppliers.DBType(),
|
|
DBWRType.DB_READ.value,
|
|
lambda s: self.user_crud.get_branding_by_session(s, sid),
|
|
)
|
|
branding = branding or {}
|
|
res.service_name = branding.get("service_name") or ""
|
|
res.logo_url = branding.get("logo_url") or ""
|
|
res.helpdesk = branding.get("helpdesk") or []
|
|
return res
|
|
|
|
async def popup_status(self, user_info: UserInfo, access_token: str) -> Res_PopupStatus:
|
|
# 유저별 팝업 숨김 상태 조회. 현재는 서비스 안내(service_info) 팝업 하나만 관리한다.
|
|
res = Res_PopupStatus()
|
|
err_type, info = await self.authenticate(user_info, access_token)
|
|
if err_type != ErrorType.SUCCESS:
|
|
res.result.SetResult(err_type)
|
|
return res
|
|
|
|
err_type, hidden = await DB_SESSION_MNG.execute_lambda(
|
|
supplier_users.DBType(),
|
|
DBWRType.DB_READ.value,
|
|
lambda s: self.user_crud.get_hide_service_info(s, uuid.UUID(info.su_id)),
|
|
)
|
|
if err_type != ErrorType.SUCCESS:
|
|
res.result.SetResult(err_type)
|
|
return res
|
|
res.service_info = hidden
|
|
return res
|
|
|
|
async def hide_popup(self, user_info: UserInfo, access_token: str, popup_type: str) -> Res_HidePopup:
|
|
# "안내 보지 않기" 영구 저장. 팝업 종류가 늘면 popup_type 분기를 추가한다.
|
|
res = Res_HidePopup()
|
|
if popup_type != "service_info":
|
|
res.result.SetResult(ErrorType.INVALID_REQUEST_DATA)
|
|
return res
|
|
|
|
err_type, info = await self.authenticate(user_info, access_token)
|
|
if err_type != ErrorType.SUCCESS:
|
|
res.result.SetResult(err_type)
|
|
return res
|
|
|
|
err_type = await DB_SESSION_MNG.execute_lambda_run(
|
|
[supplier_users.DBType()],
|
|
[lambda s: self.user_crud.set_hide_service_info(s, uuid.UUID(info.su_id))],
|
|
)
|
|
if err_type != ErrorType.SUCCESS:
|
|
res.result.SetResult(err_type)
|
|
return res
|
|
|
|
async def logout(self, user_info: UserInfo) -> Res_Logout:
|
|
# 해당 유저의 저장 토큰(access/refresh)을 모두 삭제 → 이후 보호 요청·재발급이 차단된다.
|
|
res = Res_Logout()
|
|
err_type = await DB_SESSION_MNG.execute_lambda_run(
|
|
[supplier_users.DBType()],
|
|
[lambda s: self.user_crud.delete_tokens_by_su_id(s, uuid.UUID(user_info.su_id))],
|
|
)
|
|
if err_type != ErrorType.SUCCESS:
|
|
res.result.SetResult(err_type)
|
|
return res
|
|
|
|
async def refresh_token(self, user_info: UserInfo, refresh_token: str) -> Res_RefreshToken:
|
|
# 토큰 디코드는 라우터 Depends(IsValidRefreshToken) 에서 수행됨. 여기선 su_id DB 검증 + 저장 토큰 대조 후 재발급.
|
|
res = Res_RefreshToken()
|
|
err_type, info = await self.__load_active_account(user_info.su_id)
|
|
if err_type != ErrorType.SUCCESS:
|
|
res.result.SetResult(err_type)
|
|
return res
|
|
if not await self.__verify_stored_token(info.su_id, TokenType.REFRESH.value, refresh_token):
|
|
res.result.SetResult(ErrorType.TOKEN_REVOKED) # 로그아웃/타기기 로그인으로 무효화됨
|
|
return res
|
|
|
|
new_access = CreateAccessToken(info) # DB 최신값으로 재구성한 토큰
|
|
# 단일 세션: 저장된 access 행을 새 토큰으로 갱신한다(refresh 행은 유지).
|
|
err_type = await DB_SESSION_MNG.execute_lambda_run(
|
|
[supplier_users.DBType()],
|
|
[
|
|
lambda s: self.user_crud.update_access_token(
|
|
s,
|
|
uuid.UUID(info.su_id),
|
|
{"jwt": new_access},
|
|
GTime.UTC(),
|
|
GTime.AddMinutes(jwt_token_config.access_expire_min),
|
|
)
|
|
],
|
|
)
|
|
if err_type != ErrorType.SUCCESS:
|
|
res.result.SetResult(err_type)
|
|
return res
|
|
|
|
res.access_token = new_access
|
|
return res
|