bc85c5228a
- FastAPI + SQLAlchemy(aiosqlite) 기반 SR 상태 머신 (RECEIVED → PARSED → PENDING_APPROVAL → APPROVED → IN_PROGRESS → PENDING_PM_VALIDATION → COMPLETED / FAILED_ROLLBACK) - PM 승인 워크플로우 (ApprovalFlow 테이블) - SHA-256 해시 체인 감사 로그 (위변조 방지) - AES-256-GCM 서버 자격증명 암호화 (IP/PW API 미노출) - CMDB: 기관(MOF/MOIS/MSS) + 서버 정보 관리 - 더미 데이터 자동 시딩 (6개 SR, 3개 기관, 6개 서버) - Dark-theme SPA: 대시보드 / 칸반 보드 / SR 목록 / 감사 로그 / CMDB Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
253 lines
8.0 KiB
Python
253 lines
8.0 KiB
Python
"""
|
|
ORM models + Pydantic schemas for GUARDiA ITSM
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from datetime import datetime
|
|
from enum import Enum
|
|
from typing import Any, Optional
|
|
|
|
from pydantic import BaseModel, ConfigDict
|
|
from sqlalchemy import (
|
|
Boolean, Column, DateTime, ForeignKey, Integer, String, Text, func
|
|
)
|
|
from sqlalchemy.orm import relationship
|
|
|
|
from database import Base
|
|
|
|
|
|
# ── Enums ──────────────────────────────────────────────────────────────────────
|
|
|
|
class SRStatus(str, Enum):
|
|
RECEIVED = "RECEIVED"
|
|
PARSED = "PARSED"
|
|
PENDING_APPROVAL = "PENDING_APPROVAL"
|
|
APPROVED = "APPROVED"
|
|
IN_PROGRESS = "IN_PROGRESS"
|
|
PENDING_PM_VALIDATION = "PENDING_PM_VALIDATION"
|
|
COMPLETED = "COMPLETED"
|
|
FAILED_ROLLBACK = "FAILED_ROLLBACK"
|
|
REJECTED = "REJECTED"
|
|
|
|
|
|
class SRType(str, Enum):
|
|
DEPLOY = "DEPLOY"
|
|
RESTART = "RESTART"
|
|
LOG = "LOG"
|
|
INQUIRY = "INQUIRY"
|
|
OTHER = "OTHER"
|
|
|
|
|
|
class Priority(str, Enum):
|
|
CRITICAL = "CRITICAL"
|
|
HIGH = "HIGH"
|
|
MEDIUM = "MEDIUM"
|
|
LOW = "LOW"
|
|
|
|
|
|
class ApprovalResult(str, Enum):
|
|
PENDING = "PENDING"
|
|
APPROVED = "APPROVED"
|
|
REJECTED = "REJECTED"
|
|
|
|
|
|
# ── ORM Models ─────────────────────────────────────────────────────────────────
|
|
|
|
class Institution(Base):
|
|
__tablename__ = "tb_inst_meta"
|
|
|
|
id = Column(Integer, primary_key=True, index=True)
|
|
inst_code = Column(String(20), unique=True, nullable=False, index=True)
|
|
inst_name = Column(String(100), nullable=False)
|
|
org_type = Column(String(50))
|
|
contact_pm = Column(String(100))
|
|
created_at = Column(DateTime, default=func.now())
|
|
|
|
servers = relationship("Server", back_populates="institution")
|
|
sr_requests = relationship("SRRequest", back_populates="institution")
|
|
|
|
|
|
class Server(Base):
|
|
__tablename__ = "tb_server_info"
|
|
|
|
id = Column(Integer, primary_key=True, index=True)
|
|
inst_id = Column(Integer, ForeignKey("tb_inst_meta.id"), nullable=False)
|
|
server_name = Column(String(100), nullable=False)
|
|
server_role = Column(String(20)) # WEB / WAS / DB
|
|
os_type = Column(String(50))
|
|
os_version = Column(String(50))
|
|
ip_addr = Column(String(45)) # NOT exposed in API responses
|
|
ssh_user = Column(String(50)) # NOT exposed
|
|
os_pw_enc = Column(Text) # AES-256 encrypted, NEVER exposed
|
|
port = Column(Integer, default=22)
|
|
is_active = Column(Boolean, default=True)
|
|
created_at = Column(DateTime, default=func.now())
|
|
|
|
institution = relationship("Institution", back_populates="servers")
|
|
|
|
|
|
class SRRequest(Base):
|
|
__tablename__ = "tb_sr_request"
|
|
|
|
id = Column(Integer, primary_key=True, index=True)
|
|
sr_id = Column(String(30), unique=True, nullable=False, index=True)
|
|
inst_id = Column(Integer, ForeignKey("tb_inst_meta.id"))
|
|
sr_type = Column(String(20), default=SRType.OTHER)
|
|
title = Column(String(200), nullable=False)
|
|
description = Column(Text)
|
|
status = Column(String(30), default=SRStatus.RECEIVED)
|
|
priority = Column(String(20), default=Priority.MEDIUM)
|
|
requested_by = Column(String(100))
|
|
assigned_to = Column(String(100))
|
|
target_server = Column(String(100))
|
|
created_at = Column(DateTime, default=func.now())
|
|
updated_at = Column(DateTime, default=func.now(), onupdate=func.now())
|
|
|
|
institution = relationship("Institution", back_populates="sr_requests")
|
|
approvals = relationship("ApprovalFlow", back_populates="sr_request")
|
|
audit_logs = relationship("AuditLog", back_populates="sr_request")
|
|
|
|
|
|
class ApprovalFlow(Base):
|
|
__tablename__ = "tb_approval_flow"
|
|
|
|
id = Column(Integer, primary_key=True, index=True)
|
|
sr_id = Column(String(30), ForeignKey("tb_sr_request.sr_id"), nullable=False)
|
|
approver = Column(String(100), nullable=False)
|
|
result = Column(String(20), default=ApprovalResult.PENDING)
|
|
comment = Column(Text)
|
|
decided_at = Column(DateTime)
|
|
created_at = Column(DateTime, default=func.now())
|
|
|
|
sr_request = relationship("SRRequest", back_populates="approvals")
|
|
|
|
|
|
class AuditLog(Base):
|
|
__tablename__ = "tb_audit_log"
|
|
|
|
id = Column(Integer, primary_key=True, index=True)
|
|
sr_id = Column(String(30), ForeignKey("tb_sr_request.sr_id"))
|
|
actor = Column(String(100))
|
|
action = Column(String(100), nullable=False)
|
|
detail = Column(Text)
|
|
prev_hash = Column(String(64))
|
|
log_hash = Column(String(64))
|
|
created_at = Column(DateTime, default=func.now())
|
|
|
|
sr_request = relationship("SRRequest", back_populates="audit_logs")
|
|
|
|
|
|
class OpsTask(Base):
|
|
__tablename__ = "tb_ops_task"
|
|
|
|
id = Column(Integer, primary_key=True, index=True)
|
|
sr_id = Column(String(30), ForeignKey("tb_sr_request.sr_id"))
|
|
task_name = Column(String(200), nullable=False)
|
|
task_order = Column(Integer, default=0)
|
|
status = Column(String(30), default="PENDING")
|
|
result_msg = Column(Text)
|
|
started_at = Column(DateTime)
|
|
finished_at = Column(DateTime)
|
|
created_at = Column(DateTime, default=func.now())
|
|
|
|
|
|
# ── Pydantic Schemas ───────────────────────────────────────────────────────────
|
|
|
|
class InstitutionOut(BaseModel):
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
id: int
|
|
inst_code: str
|
|
inst_name: str
|
|
org_type: Optional[str]
|
|
contact_pm: Optional[str]
|
|
|
|
|
|
class ServerOut(BaseModel):
|
|
"""Server info — sensitive fields (ip_addr, ssh_user, os_pw_enc) intentionally omitted."""
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
id: int
|
|
server_name: str
|
|
server_role: Optional[str]
|
|
os_type: Optional[str]
|
|
is_active: bool
|
|
|
|
|
|
class SRCreate(BaseModel):
|
|
title: str
|
|
description: Optional[str] = None
|
|
sr_type: SRType = SRType.OTHER
|
|
priority: Priority = Priority.MEDIUM
|
|
requested_by: str
|
|
assigned_to: Optional[str] = None
|
|
target_server: Optional[str] = None
|
|
inst_code: Optional[str] = None
|
|
|
|
|
|
class SROut(BaseModel):
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
id: int
|
|
sr_id: str
|
|
sr_type: str
|
|
title: str
|
|
description: Optional[str]
|
|
status: str
|
|
priority: str
|
|
requested_by: str
|
|
assigned_to: Optional[str]
|
|
target_server: Optional[str]
|
|
created_at: datetime
|
|
updated_at: datetime
|
|
|
|
|
|
class SRStatusUpdate(BaseModel):
|
|
status: SRStatus
|
|
actor: str
|
|
comment: Optional[str] = None
|
|
|
|
|
|
class ApprovalCreate(BaseModel):
|
|
approver: str
|
|
result: ApprovalResult
|
|
comment: Optional[str] = None
|
|
|
|
|
|
class ApprovalOut(BaseModel):
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
id: int
|
|
sr_id: str
|
|
approver: str
|
|
result: str
|
|
comment: Optional[str]
|
|
decided_at: Optional[datetime]
|
|
created_at: datetime
|
|
|
|
|
|
class AuditLogOut(BaseModel):
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
id: int
|
|
sr_id: Optional[str]
|
|
actor: Optional[str]
|
|
action: str
|
|
detail: Optional[str]
|
|
log_hash: Optional[str]
|
|
created_at: datetime
|
|
|
|
|
|
# ── Audit hash helper ──────────────────────────────────────────────────────────
|
|
|
|
def compute_log_hash(prev_hash: Optional[str], actor: str, action: str,
|
|
detail: str, ts: str) -> str:
|
|
payload = json.dumps(
|
|
{"prev": prev_hash or "", "actor": actor, "action": action,
|
|
"detail": detail, "ts": ts},
|
|
ensure_ascii=False, sort_keys=True
|
|
)
|
|
return hashlib.sha256(payload.encode()).hexdigest()
|