425 lines
17 KiB
Python
425 lines
17 KiB
Python
"""인증·조직·보안 기능의 aiomysql Raw SQL 저장소."""
|
|
|
|
import json
|
|
import secrets
|
|
import string
|
|
from datetime import datetime, timedelta
|
|
from typing import Any
|
|
|
|
import aiomysql
|
|
|
|
from config.config_db import get_db_pool
|
|
from config.config_system import OTP_VALID_MINUTES
|
|
|
|
|
|
def _company_code() -> str:
|
|
alphabet = string.ascii_uppercase + string.digits
|
|
parts = ["".join(secrets.choice(alphabet) for _ in range(3)) for _ in range(3)]
|
|
return "-".join(parts)
|
|
|
|
|
|
async def get_user_by_email(email: str) -> dict[str, Any] | None:
|
|
pool = get_db_pool()
|
|
async with pool.acquire() as connection, connection.cursor(aiomysql.DictCursor) as cursor:
|
|
await cursor.execute(
|
|
"SELECT * FROM users WHERE email = %s AND deleted_at IS NULL", (email,)
|
|
)
|
|
return await cursor.fetchone()
|
|
|
|
|
|
async def create_registration(data: dict[str, Any], password_hash: str) -> int:
|
|
pool = get_db_pool()
|
|
async with pool.acquire() as connection, connection.cursor() as cursor:
|
|
try:
|
|
await connection.begin()
|
|
await cursor.execute(
|
|
"""INSERT INTO users
|
|
(email, password_hash, name, position, phone, role, is_master, status)
|
|
VALUES (%s, %s, %s, %s, %s, 'USER', FALSE, 'PENDING_EMAIL')""",
|
|
(
|
|
data["email"],
|
|
password_hash,
|
|
data["name"],
|
|
data.get("position"),
|
|
data.get("phone"),
|
|
),
|
|
)
|
|
user_id = cursor.lastrowid
|
|
await cursor.execute(
|
|
"""INSERT INTO user_consents
|
|
(user_id, terms_version, terms_agreed, privacy_agreed, marketing_agreed)
|
|
VALUES (%s, %s, %s, %s, %s)""",
|
|
(
|
|
user_id,
|
|
data["terms_version"],
|
|
data["terms_agreed"],
|
|
data["privacy_agreed"],
|
|
data["marketing_agreed"],
|
|
),
|
|
)
|
|
await connection.commit()
|
|
return user_id
|
|
except Exception:
|
|
await connection.rollback()
|
|
raise
|
|
|
|
|
|
async def refresh_pending_registration(
|
|
user_id: int, data: dict[str, Any], password_hash: str
|
|
) -> None:
|
|
"""미인증(PENDING_EMAIL) 계정의 정보를 최신 가입 요청으로 갱신한다.
|
|
|
|
같은 이메일로 재가입을 시도하면 기존 미인증 계정을 덮어써서
|
|
새 비밀번호·이름·약관 동의로 갱신하고 새 OTP를 받을 수 있게 한다.
|
|
"""
|
|
pool = get_db_pool()
|
|
async with pool.acquire() as connection, connection.cursor() as cursor:
|
|
try:
|
|
await connection.begin()
|
|
await cursor.execute(
|
|
"""UPDATE users
|
|
SET password_hash = %s, name = %s, position = %s, phone = %s
|
|
WHERE id = %s AND status = 'PENDING_EMAIL'""",
|
|
(
|
|
password_hash,
|
|
data["name"],
|
|
data.get("position"),
|
|
data.get("phone"),
|
|
user_id,
|
|
),
|
|
)
|
|
await cursor.execute(
|
|
"""INSERT INTO user_consents
|
|
(user_id, terms_version, terms_agreed, privacy_agreed, marketing_agreed)
|
|
VALUES (%s, %s, %s, %s, %s)
|
|
ON DUPLICATE KEY UPDATE
|
|
terms_agreed = VALUES(terms_agreed),
|
|
privacy_agreed = VALUES(privacy_agreed),
|
|
marketing_agreed = VALUES(marketing_agreed)""",
|
|
(
|
|
user_id,
|
|
data["terms_version"],
|
|
data["terms_agreed"],
|
|
data["privacy_agreed"],
|
|
data["marketing_agreed"],
|
|
),
|
|
)
|
|
await connection.commit()
|
|
except Exception:
|
|
await connection.rollback()
|
|
raise
|
|
|
|
|
|
async def replace_otp(user_id: int, purpose: str, otp_hash: str) -> None:
|
|
expires_at = datetime.utcnow() + timedelta(minutes=OTP_VALID_MINUTES)
|
|
pool = get_db_pool()
|
|
async with pool.acquire() as connection, connection.cursor() as cursor:
|
|
await cursor.execute(
|
|
"""UPDATE email_otps SET consumed_at = CURRENT_TIMESTAMP
|
|
WHERE user_id = %s AND purpose = %s AND consumed_at IS NULL""",
|
|
(user_id, purpose),
|
|
)
|
|
await cursor.execute(
|
|
"""INSERT INTO email_otps (user_id, purpose, otp_hash, expires_at)
|
|
VALUES (%s, %s, %s, %s)""",
|
|
(user_id, purpose, otp_hash, expires_at),
|
|
)
|
|
await connection.commit()
|
|
|
|
|
|
async def get_active_otp(user_id: int, purpose: str) -> dict[str, Any] | None:
|
|
pool = get_db_pool()
|
|
async with pool.acquire() as connection, connection.cursor(aiomysql.DictCursor) as cursor:
|
|
await cursor.execute(
|
|
"""SELECT id, otp_hash, expires_at FROM email_otps
|
|
WHERE user_id = %s AND purpose = %s AND consumed_at IS NULL
|
|
ORDER BY created_at DESC LIMIT 1""",
|
|
(user_id, purpose),
|
|
)
|
|
return await cursor.fetchone()
|
|
|
|
|
|
async def consume_otp(otp_id: int) -> None:
|
|
pool = get_db_pool()
|
|
async with pool.acquire() as connection, connection.cursor() as cursor:
|
|
await cursor.execute(
|
|
"UPDATE email_otps SET consumed_at = CURRENT_TIMESTAMP WHERE id = %s", (otp_id,)
|
|
)
|
|
await connection.commit()
|
|
|
|
|
|
async def complete_registration(user_id: int, is_master: bool) -> None:
|
|
pool = get_db_pool()
|
|
async with pool.acquire() as connection, connection.cursor() as cursor:
|
|
await cursor.execute(
|
|
"""UPDATE users SET status = 'NO_COMPANY', last_email_verified_at = CURRENT_TIMESTAMP,
|
|
auth_expires_at = DATE_ADD(CURRENT_TIMESTAMP, INTERVAL 3 MONTH)
|
|
WHERE id = %s""",
|
|
(user_id,),
|
|
)
|
|
await connection.commit()
|
|
|
|
|
|
async def company_master_email(company_id: int) -> str | None:
|
|
pool = get_db_pool()
|
|
async with pool.acquire() as connection, connection.cursor() as cursor:
|
|
await cursor.execute(
|
|
"""SELECT u.email FROM companies c JOIN users u ON u.id = c.master_user_id
|
|
WHERE c.id = %s AND c.status = 'ACTIVE'""",
|
|
(company_id,),
|
|
)
|
|
row = await cursor.fetchone()
|
|
return row[0] if row else None
|
|
|
|
|
|
async def search_companies(query: str) -> list[dict[str, Any]]:
|
|
pool = get_db_pool()
|
|
async with pool.acquire() as connection, connection.cursor(aiomysql.DictCursor) as cursor:
|
|
await cursor.execute(
|
|
"""SELECT id, name FROM companies
|
|
WHERE status = 'ACTIVE' AND deleted_at IS NULL AND name LIKE %s
|
|
ORDER BY name LIMIT 20""",
|
|
(f"%{query}%",),
|
|
)
|
|
return list(await cursor.fetchall())
|
|
|
|
|
|
async def record_login(
|
|
user_id: int | None, email: str, status: str, reason: str | None, user_agent: str
|
|
) -> None:
|
|
pool = get_db_pool()
|
|
async with pool.acquire() as connection, connection.cursor() as cursor:
|
|
await cursor.execute(
|
|
"""INSERT INTO login_logs (user_id, email, status, failure_reason, user_agent)
|
|
VALUES (%s, %s, %s, %s, %s)""",
|
|
(user_id, email, status, reason, user_agent),
|
|
)
|
|
await connection.commit()
|
|
|
|
|
|
async def record_failed_login(user_id: int) -> int:
|
|
pool = get_db_pool()
|
|
async with pool.acquire() as connection, connection.cursor() as cursor:
|
|
await cursor.execute(
|
|
"""UPDATE users SET login_failures = login_failures + 1,
|
|
last_failed_at = CURRENT_TIMESTAMP,
|
|
account_locked_until = CASE WHEN login_failures + 1 >= 5
|
|
THEN DATE_ADD(CURRENT_TIMESTAMP, INTERVAL 15 MINUTE)
|
|
ELSE account_locked_until END
|
|
WHERE id = %s""",
|
|
(user_id,),
|
|
)
|
|
await cursor.execute("SELECT login_failures FROM users WHERE id = %s", (user_id,))
|
|
failures = (await cursor.fetchone())[0]
|
|
await connection.commit()
|
|
return failures
|
|
|
|
|
|
async def clear_login_failures(user_id: int) -> None:
|
|
pool = get_db_pool()
|
|
async with pool.acquire() as connection, connection.cursor() as cursor:
|
|
await cursor.execute(
|
|
"""UPDATE users SET login_failures = 0, last_failed_at = NULL,
|
|
account_locked_until = NULL WHERE id = %s""",
|
|
(user_id,),
|
|
)
|
|
await connection.commit()
|
|
|
|
|
|
async def has_known_browser(user_id: int, user_agent_hash: str) -> bool:
|
|
pool = get_db_pool()
|
|
async with pool.acquire() as connection, connection.cursor() as cursor:
|
|
await cursor.execute(
|
|
"SELECT 1 FROM trusted_devices WHERE user_id = %s AND user_agent_hash = %s LIMIT 1",
|
|
(user_id, user_agent_hash),
|
|
)
|
|
return await cursor.fetchone() is not None
|
|
|
|
|
|
async def trust_browser(user_id: int, user_agent_hash: str) -> None:
|
|
pool = get_db_pool()
|
|
async with pool.acquire() as connection, connection.cursor() as cursor:
|
|
await cursor.execute(
|
|
"""INSERT INTO trusted_devices (user_id, user_agent_hash)
|
|
VALUES (%s, %s) ON DUPLICATE KEY UPDATE last_used_at = CURRENT_TIMESTAMP""",
|
|
(user_id, user_agent_hash),
|
|
)
|
|
await connection.commit()
|
|
|
|
|
|
async def delete_session(session_id: str) -> None:
|
|
pool = get_db_pool()
|
|
async with pool.acquire() as connection, connection.cursor() as cursor:
|
|
await cursor.execute("DELETE FROM sessions WHERE id = %s", (session_id,))
|
|
await connection.commit()
|
|
|
|
|
|
async def delete_user_sessions(user_id: int) -> None:
|
|
pool = get_db_pool()
|
|
async with pool.acquire() as connection, connection.cursor() as cursor:
|
|
await cursor.execute("DELETE FROM sessions WHERE user_id = %s", (user_id,))
|
|
await connection.commit()
|
|
|
|
|
|
async def change_password(user_id: int, password_hash: str, logout_all: bool) -> None:
|
|
pool = get_db_pool()
|
|
async with pool.acquire() as connection, connection.cursor() as cursor:
|
|
await cursor.execute(
|
|
"""UPDATE users SET password_hash = %s, last_email_verified_at = NULL
|
|
WHERE id = %s""",
|
|
(password_hash, user_id),
|
|
)
|
|
if logout_all:
|
|
await cursor.execute("DELETE FROM sessions WHERE user_id = %s", (user_id,))
|
|
await connection.commit()
|
|
|
|
|
|
async def list_join_requests(company_id: int) -> list[dict[str, Any]]:
|
|
pool = get_db_pool()
|
|
async with pool.acquire() as connection, connection.cursor(aiomysql.DictCursor) as cursor:
|
|
await cursor.execute(
|
|
"""SELECT jr.id, jr.status, jr.requested_at, u.id AS user_id, u.name, u.email
|
|
FROM join_requests jr JOIN users u ON u.id = jr.user_id
|
|
WHERE jr.company_id = %s ORDER BY jr.requested_at DESC""",
|
|
(company_id,),
|
|
)
|
|
return list(await cursor.fetchall())
|
|
|
|
|
|
async def decide_join_request(
|
|
request_id: int, company_id: int, reviewer_id: int, approved: bool, comment: str | None
|
|
) -> dict[str, Any] | None:
|
|
pool = get_db_pool()
|
|
async with pool.acquire() as connection, connection.cursor(aiomysql.DictCursor) as cursor:
|
|
try:
|
|
await connection.begin()
|
|
await cursor.execute(
|
|
"""SELECT user_id FROM join_requests
|
|
WHERE id = %s AND company_id = %s AND status = 'PENDING'
|
|
FOR UPDATE""",
|
|
(request_id, company_id),
|
|
)
|
|
request_row = await cursor.fetchone()
|
|
if not request_row:
|
|
await connection.rollback()
|
|
return None
|
|
decision = "APPROVED" if approved else "REJECTED"
|
|
user_status = "ACTIVE" if approved else "REJECTED"
|
|
await cursor.execute(
|
|
"""UPDATE join_requests SET status = %s, reviewed_at = CURRENT_TIMESTAMP,
|
|
reviewed_by = %s, review_comment = %s WHERE id = %s""",
|
|
(decision, reviewer_id, comment, request_id),
|
|
)
|
|
await cursor.execute(
|
|
"UPDATE users SET status = %s WHERE id = %s", (user_status, request_row["user_id"])
|
|
)
|
|
await cursor.execute(
|
|
"SELECT id, email, name FROM users WHERE id = %s", (request_row["user_id"],)
|
|
)
|
|
user = await cursor.fetchone()
|
|
await connection.commit()
|
|
return user
|
|
except Exception:
|
|
await connection.rollback()
|
|
raise
|
|
|
|
|
|
async def list_company_members(company_id: int) -> list[dict[str, Any]]:
|
|
pool = get_db_pool()
|
|
async with pool.acquire() as connection, connection.cursor(aiomysql.DictCursor) as cursor:
|
|
await cursor.execute(
|
|
"""SELECT id, name, email, role, status, last_email_verified_at, created_at
|
|
FROM users WHERE company_id = %s AND deleted_at IS NULL ORDER BY name""",
|
|
(company_id,),
|
|
)
|
|
return list(await cursor.fetchall())
|
|
|
|
|
|
async def remove_member(user_id: int, company_id: int) -> bool:
|
|
pool = get_db_pool()
|
|
async with pool.acquire() as connection, connection.cursor() as cursor:
|
|
await cursor.execute(
|
|
"""UPDATE users SET status = 'INACTIVE' WHERE id = %s AND company_id = %s
|
|
AND is_master = FALSE""",
|
|
(user_id, company_id),
|
|
)
|
|
changed = cursor.rowcount > 0
|
|
if changed:
|
|
await cursor.execute("DELETE FROM sessions WHERE user_id = %s", (user_id,))
|
|
await connection.commit()
|
|
return changed
|
|
|
|
|
|
async def master_dashboard(company_id: int) -> dict[str, Any]:
|
|
pool = get_db_pool()
|
|
async with pool.acquire() as connection, connection.cursor(aiomysql.DictCursor) as cursor:
|
|
await cursor.execute(
|
|
"""SELECT c.id, c.name, c.subscription_status,
|
|
COUNT(DISTINCT u.id) AS member_count,
|
|
SUM(u.status = 'ACTIVE') AS active_member_count,
|
|
COUNT(DISTINCT jr.id) AS pending_request_count
|
|
FROM companies c LEFT JOIN users u ON u.company_id = c.id
|
|
LEFT JOIN join_requests jr ON jr.company_id = c.id AND jr.status = 'PENDING'
|
|
WHERE c.id = %s GROUP BY c.id""",
|
|
(company_id,),
|
|
)
|
|
return await cursor.fetchone() or {}
|
|
|
|
|
|
async def admin_dashboard() -> dict[str, Any]:
|
|
pool = get_db_pool()
|
|
async with pool.acquire() as connection, connection.cursor(aiomysql.DictCursor) as cursor:
|
|
await cursor.execute(
|
|
"""SELECT
|
|
(SELECT COUNT(*) FROM users WHERE status = 'ACTIVE') AS active_users,
|
|
(SELECT COUNT(*) FROM companies WHERE status = 'ACTIVE') AS active_companies,
|
|
(SELECT COUNT(*) FROM sessions
|
|
WHERE expires_at > CURRENT_TIMESTAMP) AS active_sessions,
|
|
(SELECT COUNT(*) FROM support_requests
|
|
WHERE status = 'NEW') AS new_support_requests"""
|
|
)
|
|
return await cursor.fetchone() or {}
|
|
|
|
|
|
async def admin_list(table: str) -> list[dict[str, Any]]:
|
|
queries = {
|
|
"companies": """SELECT id, name, status, subscription_status, created_at
|
|
FROM companies ORDER BY created_at DESC""",
|
|
"users": """SELECT id, email, name, company_id, role, status, created_at
|
|
FROM users ORDER BY created_at DESC""",
|
|
"support": "SELECT * FROM support_requests ORDER BY created_at DESC",
|
|
}
|
|
if table not in queries:
|
|
raise ValueError("지원하지 않는 관리자 목록입니다.")
|
|
pool = get_db_pool()
|
|
async with pool.acquire() as connection, connection.cursor(aiomysql.DictCursor) as cursor:
|
|
await cursor.execute(queries[table])
|
|
return list(await cursor.fetchall())
|
|
|
|
|
|
async def admin_update_status(
|
|
admin_id: int, entity: str, target_id: int, status: str, details: dict[str, Any]
|
|
) -> bool:
|
|
allowed = {
|
|
"company": ("companies", {"ACTIVE", "INACTIVE", "SUSPENDED"}),
|
|
"user": ("users", {"ACTIVE", "INACTIVE", "REJECTED"}),
|
|
"support": ("support_requests", {"NEW", "REVIEWED", "RESOLVED", "DELETED"}),
|
|
}
|
|
if entity not in allowed or status not in allowed[entity][1]:
|
|
return False
|
|
table = allowed[entity][0]
|
|
pool = get_db_pool()
|
|
async with pool.acquire() as connection, connection.cursor() as cursor:
|
|
await connection.begin()
|
|
await cursor.execute(f"UPDATE {table} SET status = %s WHERE id = %s", (status, target_id))
|
|
changed = cursor.rowcount > 0
|
|
if entity == "user" and status != "ACTIVE":
|
|
await cursor.execute("DELETE FROM sessions WHERE user_id = %s", (target_id,))
|
|
await cursor.execute(
|
|
"""INSERT INTO system_admin_logs (admin_user_id, action_type, target_id, details)
|
|
VALUES (%s, %s, %s, %s)""",
|
|
(admin_id, f"{entity.upper()}_STATUS", str(target_id), json.dumps(details)),
|
|
)
|
|
await connection.commit()
|
|
return changed
|