hyungi
74876b674c
PR-Infra-Sec-1H Phase 0 audit 에서 DS jwt invalidation 정책 부재 확정. password rotation 으로 구 365d JWT (voice-memo-bot 등) invalidate 안 되는 hard gate STOP 진입 → 선행 PR 분리. - migration 269: users.password_changed_at timestamptz NULL (legacy 호환) - create_access_token / create_refresh_token: payload 에 iat (int 초) 추가 - verify_password_changed_at helper: int(password_changed_at.timestamp()) > int(iat) 시 401 - get_current_user + refresh_token route: verify helper 호출 - change_password / setup signup / seed_admin INSERT+UPDATE: password_changed_at 갱신 NULL = 검증 skip (migration 직후 운영 영향 0). 첫 password 변경 후만 iat 검증 활성. Sec-1H 의 G-token-old hard gate 통과 path 확보.
79 lines
2.5 KiB
Python
79 lines
2.5 KiB
Python
"""초기 관리자 계정 생성 스크립트
|
|
|
|
사용법:
|
|
# Docker 컨테이너 내부에서 실행
|
|
docker compose exec fastapi python /app/scripts/seed_admin.py
|
|
|
|
# 로컬에서 실행 (DATABASE_URL 환경변수 필요)
|
|
python scripts/seed_admin.py
|
|
"""
|
|
|
|
import asyncio
|
|
import getpass
|
|
import os
|
|
import sys
|
|
|
|
# 프로젝트 루트의 app/ 디렉토리를 경로에 추가
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "app"))
|
|
|
|
from sqlalchemy import select, text
|
|
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
|
|
|
|
from core.auth import hash_password
|
|
|
|
|
|
async def seed_admin():
|
|
database_url = os.getenv(
|
|
"DATABASE_URL",
|
|
"postgresql+asyncpg://pkm:pkm@localhost:5432/pkm",
|
|
)
|
|
|
|
engine = create_async_engine(database_url)
|
|
async_session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
|
|
|
|
print("=== hyungi_Document_Server 관리자 계정 생성 ===\n")
|
|
|
|
username = input("관리자 아이디 [admin]: ").strip() or "admin"
|
|
password = getpass.getpass("비밀번호: ")
|
|
if not password:
|
|
print("비밀번호는 필수입니다.")
|
|
return
|
|
|
|
password_confirm = getpass.getpass("비밀번호 확인: ")
|
|
if password != password_confirm:
|
|
print("비밀번호가 일치하지 않습니다.")
|
|
return
|
|
|
|
password_hash = hash_password(password)
|
|
|
|
async with async_session() as session:
|
|
# 이미 존재하는지 확인
|
|
result = await session.execute(
|
|
text("SELECT id FROM users WHERE username = :username"),
|
|
{"username": username},
|
|
)
|
|
if result.scalar_one_or_none():
|
|
print(f"'{username}' 계정이 이미 존재합니다. 비밀번호를 업데이트합니다.")
|
|
await session.execute(
|
|
text("UPDATE users SET password_hash = :hash, password_changed_at = NOW() WHERE username = :username"),
|
|
{"hash": password_hash, "username": username},
|
|
)
|
|
else:
|
|
await session.execute(
|
|
text(
|
|
"INSERT INTO users (username, password_hash, is_active, password_changed_at) "
|
|
"VALUES (:username, :hash, TRUE, NOW())"
|
|
),
|
|
{"username": username, "hash": password_hash},
|
|
)
|
|
print(f"'{username}' 계정이 생성되었습니다.")
|
|
|
|
await session.commit()
|
|
|
|
await engine.dispose()
|
|
print("\n완료!")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(seed_admin())
|