G-1: 메신저 Webhook Relay + _send_to_room 실제 httpx 호출 구현 G-2: POST /api/tasks/bulk SR 대량작업 엔드포인트 (최대 100건) G-3: 라이선스 만료 알림 스케줄러 (매일 09:00 KST) G-4: 체험판 upgrade_banner 필드 + license.py 배너 로직 G-5: core/auto_rca.py + incidents/problem auto-rca 엔드포인트 G-6: core/deploy_impact.py + vibe impact-analysis 엔드포인트 G-7: core/ticket_classifier.py + SR 생성 시 AI 분류 + ai-suggestion API G-8: VulnPatchRecord 모델 + vuln_scan 패치추적 4개 엔드포인트 G-9: core/jira_sync.py + gateway Jira/Confluence 연동 엔드포인트 G-10: core/push_notify.py + routers/push.py + PushSubscription 모델 G-11: approvals 다중승인 (위임/서명/기한초과/마감연장) G-12: alembic.ini + migrations/ + cicd/migrate_to_postgres.sh 하네스: guardia-orchestrator 확장기능 Phase 반영 봇명령어: /sr /status /license /bulk 슬래시 명령어 추가 설치스크립트: setup/ (Ubuntu, CentOS, RHEL, Windows) --test 옵션 포함 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
952 lines
26 KiB
Markdown
952 lines
26 KiB
Markdown
# GUARDiA ITSM + Messenger — 개발자 지침서
|
|
|
|
**문서 버전**: 1.0
|
|
**작성일**: 2026-05-25
|
|
**대상**: GUARDiA 플랫폼 개발자
|
|
|
|
---
|
|
|
|
## 목차
|
|
|
|
1. [개발 환경 설정](#1-개발-환경-설정)
|
|
2. [프로젝트 구조](#2-프로젝트-구조)
|
|
3. [코딩 컨벤션](#3-코딩-컨벤션)
|
|
4. [API 개발 가이드](#4-api-개발-가이드)
|
|
5. [데이터베이스 가이드](#5-데이터베이스-가이드)
|
|
6. [보안 필수 규칙](#6-보안-필수-규칙)
|
|
7. [테스트 작성 가이드](#7-테스트-작성-가이드)
|
|
8. [Messenger Bot 개발 가이드](#8-messenger-bot-개발-가이드)
|
|
9. [스케줄러 확장 가이드](#9-스케줄러-확장-가이드)
|
|
10. [배포 및 CI/CD](#10-배포-및-cicd)
|
|
11. [자주 하는 실수 및 주의사항](#11-자주-하는-실수-및-주의사항)
|
|
|
|
---
|
|
|
|
## 1. 개발 환경 설정
|
|
|
|
### 1.1 필수 도구
|
|
|
|
```
|
|
Python 3.11+
|
|
Git
|
|
VS Code (권장) 또는 PyCharm
|
|
SQLite Browser (DB 확인용)
|
|
httpie 또는 Postman (API 테스트)
|
|
```
|
|
|
|
### 1.2 로컬 개발 환경 구성
|
|
|
|
```bash
|
|
# 1. 레포지토리 클론
|
|
git clone <repo-url>
|
|
cd GUARDiA/itsm
|
|
|
|
# 2. 가상환경 생성 (Python 3.11 사용)
|
|
python -m venv .venv
|
|
|
|
# Windows
|
|
.venv\Scripts\activate
|
|
# Linux/Mac
|
|
source .venv/bin/activate
|
|
|
|
# 3. 의존성 설치
|
|
pip install -r requirements.txt
|
|
pip install -r requirements-dev.txt # 개발/테스트용
|
|
|
|
# 4. 환경 변수 설정
|
|
cp .env.example .env
|
|
# .env 파일 편집 (SECRET_KEY, DB_PATH 등)
|
|
|
|
# 5. 앱 실행
|
|
uvicorn main:app --reload --port 8000
|
|
|
|
# 6. API 문서 확인
|
|
# http://localhost:8000/docs
|
|
```
|
|
|
|
### 1.3 .env 파일 설정
|
|
|
|
```ini
|
|
# .env.example
|
|
SECRET_KEY=your-256-bit-secret-key-here-change-in-production
|
|
ALGORITHM=HS256
|
|
ACCESS_TOKEN_EXPIRE_MINUTES=480
|
|
DB_URL=sqlite+aiosqlite:///./guardia_itsm.db
|
|
|
|
# 암호화 키 (AES-256-GCM, 32바이트 hex)
|
|
ENCRYPTION_KEY=0000000000000000000000000000000000000000000000000000000000000000
|
|
|
|
# 라이선스 마스터 키 (64자리 hex, 32바이트)
|
|
# 생성: python -c "import secrets; print(secrets.token_hex(32))"
|
|
# 운영 환경에서는 반드시 실제 값으로 교체 — 분실 시 기발급 라이선스 전부 무효화됨
|
|
GUARDIA_LICENSE_KEY=0000000000000000000000000000000000000000000000000000000000000000
|
|
|
|
# Messenger 연동
|
|
MESSENGER_BASE_URL=http://localhost:8002
|
|
MESSENGER_BOT_TOKEN=dev-token
|
|
|
|
# SSH 실행 타임아웃 (초)
|
|
SSH_TIMEOUT=30
|
|
|
|
# 파일 업로드 경로
|
|
UPLOAD_ROOT=./uploads
|
|
MAX_FILE_SIZE_MB=10
|
|
```
|
|
|
|
> **주의**: `.env` 파일은 절대 Git에 커밋하지 마세요. `.gitignore`에 포함되어 있습니다.
|
|
|
|
### 1.4 VS Code 권장 설정
|
|
|
|
`.vscode/settings.json`:
|
|
```json
|
|
{
|
|
"python.defaultInterpreterPath": ".venv/Scripts/python.exe",
|
|
"editor.formatOnSave": true,
|
|
"python.formatting.provider": "black",
|
|
"python.linting.enabled": true,
|
|
"python.linting.pylintEnabled": false,
|
|
"python.linting.flake8Enabled": true,
|
|
"python.linting.flake8Args": ["--max-line-length=120"]
|
|
}
|
|
```
|
|
|
|
---
|
|
|
|
## 2. 프로젝트 구조
|
|
|
|
```
|
|
C:\GUARDiA\itsm\
|
|
├── main.py # FastAPI 앱 진입점
|
|
├── database.py # DB 엔진, SessionLocal, get_db
|
|
├── models.py # SQLAlchemy ORM 모델 (전체)
|
|
├── schemas.py # Pydantic 스키마 (전체)
|
|
├── requirements.txt # 운영 의존성
|
|
├── requirements-dev.txt # 개발/테스트 의존성
|
|
├── .env # 환경 변수 (git 제외)
|
|
│
|
|
├── core\
|
|
│ ├── seed.py # 초기 데이터 시드
|
|
│ ├── scheduler.py # APScheduler 백그라운드 작업
|
|
│ └── security.py # JWT, 비밀번호 해싱
|
|
│
|
|
├── routers\
|
|
│ ├── auth.py # 인증/권한
|
|
│ ├── tasks.py # SR 처리
|
|
│ ├── approvals.py # 결재
|
|
│ ├── ssl_manager.py # SSL 인증서 관리
|
|
│ ├── pm.py # 정기점검
|
|
│ ├── incidents.py # 장애 관리
|
|
│ ├── oncall.py # 온콜/당직
|
|
│ ├── batch.py # 배치 작업
|
|
│ └── ... # 기타 라우터
|
|
│
|
|
├── utils\
|
|
│ ├── crypto.py # AES-256-GCM 암/복호화
|
|
│ └── ssh_runner.py # SSH 명령 실행
|
|
│
|
|
├── static\ # 프론트엔드 (HTML/CSS/JS)
|
|
├── uploads\ # 업로드 파일 저장소
|
|
│ ├── sr_files\
|
|
│ └── workspaces\
|
|
│
|
|
├── scripts\sm\ssl\
|
|
│ └── ssl_expiry_check.sh # 배포용 SSL 점검 스크립트
|
|
│
|
|
└── tests\
|
|
├── conftest.py
|
|
├── unit\
|
|
└── integration\
|
|
|
|
C:\GUARDiA\messenger\
|
|
├── main.py # Messenger 서버
|
|
├── models\ # DB 모델
|
|
├── core\ # 봇 로직
|
|
└── routers\ # API 라우터
|
|
```
|
|
|
|
---
|
|
|
|
## 3. 코딩 컨벤션
|
|
|
|
### 3.1 Python 스타일
|
|
|
|
```python
|
|
# 파일 맨 위: 임포트 순서 (isort 준수)
|
|
# 1) 표준 라이브러리
|
|
import os
|
|
import json
|
|
from datetime import datetime, timedelta
|
|
|
|
# 2) 서드파티
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
# 3) 로컬 모듈
|
|
from database import get_db
|
|
from models import SRRequest
|
|
from schemas import SRCreate, SROut
|
|
|
|
# 줄 최대 길이: 120자
|
|
# 들여쓰기: 4 space (탭 금지)
|
|
# 문자열: 큰따옴표(") 사용
|
|
```
|
|
|
|
### 3.2 라우터 파일 구조 패턴
|
|
|
|
새 라우터를 만들 때 반드시 이 구조를 따르세요:
|
|
|
|
```python
|
|
# routers/example.py
|
|
from datetime import datetime
|
|
from typing import Optional
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from database import get_db
|
|
from models import ExampleModel
|
|
from schemas import ExampleCreate, ExampleOut, ExampleUpdate
|
|
from core.security import get_current_user
|
|
from models import User, Role
|
|
|
|
router = APIRouter(prefix="/example", tags=["example"])
|
|
|
|
|
|
def _require_role(*roles: Role):
|
|
"""역할 기반 접근 제어 의존성"""
|
|
async def checker(current_user: User = Depends(get_current_user)):
|
|
if current_user.role not in roles:
|
|
raise HTTPException(status_code=403, detail="권한이 없습니다.")
|
|
return current_user
|
|
return checker
|
|
|
|
|
|
@router.get("/", response_model=list[ExampleOut])
|
|
async def list_examples(
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
result = await db.execute(select(ExampleModel).order_by(ExampleModel.created_at.desc()))
|
|
return result.scalars().all()
|
|
|
|
|
|
@router.post("/", response_model=ExampleOut, status_code=status.HTTP_201_CREATED)
|
|
async def create_example(
|
|
body: ExampleCreate,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(_require_role(Role.ADMIN, Role.PM)),
|
|
):
|
|
obj = ExampleModel(**body.model_dump(), created_by=current_user.id)
|
|
db.add(obj)
|
|
await db.commit()
|
|
await db.refresh(obj)
|
|
return obj
|
|
```
|
|
|
|
### 3.3 모델 정의 패턴
|
|
|
|
```python
|
|
# models.py 에 추가할 때
|
|
class ExampleModel(Base):
|
|
__tablename__ = "tb_example"
|
|
|
|
id = Column(Integer, primary_key=True, autoincrement=True)
|
|
title = Column(String(200), nullable=False)
|
|
description = Column(Text, nullable=True)
|
|
is_active = Column(Boolean, default=True, nullable=False)
|
|
created_by = Column(Integer, ForeignKey("tb_user.id"), nullable=True)
|
|
created_at = Column(DateTime, default=datetime.utcnow, nullable=False)
|
|
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
|
|
```
|
|
|
|
### 3.4 스키마 정의 패턴
|
|
|
|
```python
|
|
# schemas.py 에 추가할 때
|
|
class ExampleCreate(BaseModel):
|
|
title: str = Field(..., min_length=1, max_length=200)
|
|
description: Optional[str] = None
|
|
|
|
class ExampleUpdate(BaseModel):
|
|
title: Optional[str] = Field(None, min_length=1, max_length=200)
|
|
description: Optional[str] = None
|
|
|
|
class ExampleOut(BaseModel):
|
|
id: int
|
|
title: str
|
|
description: Optional[str]
|
|
is_active: bool
|
|
created_at: datetime
|
|
|
|
model_config = ConfigDict(from_attributes=True)
|
|
```
|
|
|
|
### 3.5 명명 규칙
|
|
|
|
| 대상 | 규칙 | 예시 |
|
|
|------|------|------|
|
|
| 파일명 | snake_case | `ssl_manager.py` |
|
|
| 클래스명 | PascalCase | `SslDomain`, `PmSchedule` |
|
|
| 함수/변수 | snake_case | `get_ssl_domains()`, `days_left` |
|
|
| DB 테이블 | `tb_` 접두사 | `tb_ssl_domain` |
|
|
| 상수 | UPPER_SNAKE | `MAX_FILE_SIZE`, `SSL_WARN_DAYS` |
|
|
| 비공개 함수 | `_` 접두사 | `_notify_incident()` |
|
|
| API 경로 | kebab-case | `/ssl-manager/domains` |
|
|
|
|
---
|
|
|
|
## 4. API 개발 가이드
|
|
|
|
### 4.1 응답 형식 규칙
|
|
|
|
```python
|
|
# 성공 응답: HTTP 상태 코드로 구분
|
|
# 200 OK — 조회, 수정
|
|
# 201 Created — 생성
|
|
# 202 Accepted — 비동기 실행 (배치, 백그라운드)
|
|
# 204 No Content — 삭제
|
|
|
|
# 에러 응답: 반드시 이 형식 사용
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="사용자 친화적인 한국어 메시지"
|
|
)
|
|
|
|
# 절대 금지: 스택트레이스 노출
|
|
# raise Exception(traceback.format_exc()) # ❌
|
|
```
|
|
|
|
### 4.2 페이징 구현 패턴
|
|
|
|
```python
|
|
from fastapi import Query
|
|
|
|
@router.get("/", response_model=dict)
|
|
async def list_items(
|
|
page: int = Query(1, ge=1),
|
|
size: int = Query(20, ge=1, le=100),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
offset = (page - 1) * size
|
|
count_q = select(func.count()).select_from(Item)
|
|
total = (await db.execute(count_q)).scalar()
|
|
|
|
items_q = select(Item).offset(offset).limit(size).order_by(Item.created_at.desc())
|
|
items = (await db.execute(items_q)).scalars().all()
|
|
|
|
return {
|
|
"items": items,
|
|
"total": total,
|
|
"page": page,
|
|
"size": size,
|
|
"pages": (total + size - 1) // size,
|
|
}
|
|
```
|
|
|
|
### 4.3 비동기 DB 쿼리 패턴
|
|
|
|
```python
|
|
# ✅ 올바른 패턴: asyncio 방식
|
|
async def get_item(item_id: int, db: AsyncSession) -> Item:
|
|
result = await db.execute(select(Item).where(Item.id == item_id))
|
|
item = result.scalar_one_or_none()
|
|
if item is None:
|
|
raise HTTPException(status_code=404, detail="항목을 찾을 수 없습니다.")
|
|
return item
|
|
|
|
# ✅ 여러 조건 필터
|
|
query = select(Item).where(
|
|
Item.is_active == True,
|
|
Item.category == category,
|
|
).order_by(Item.created_at.desc())
|
|
|
|
# ✅ JOIN
|
|
query = select(Item).join(User, Item.created_by == User.id)
|
|
|
|
# ❌ 잘못된 패턴: 동기 방식 사용 금지
|
|
# db.query(Item).filter(Item.id == item_id).first() # ❌
|
|
```
|
|
|
|
### 4.4 백그라운드 작업 패턴
|
|
|
|
```python
|
|
from fastapi import BackgroundTasks
|
|
|
|
@router.post("/{id}/run", status_code=202)
|
|
async def run_job(
|
|
id: int,
|
|
background_tasks: BackgroundTasks,
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
job = await _get_job(id, db)
|
|
background_tasks.add_task(_execute_job, job.id)
|
|
return {"message": "실행이 요청되었습니다.", "job_id": id}
|
|
|
|
|
|
async def _execute_job(job_id: int):
|
|
# 중요: 새 DB 세션 생성 (request-scope 세션 재사용 금지!)
|
|
from database import SessionLocal
|
|
async with SessionLocal() as db:
|
|
job = await _get_job(job_id, db)
|
|
# ... 실행 로직
|
|
await db.commit()
|
|
```
|
|
|
|
### 4.5 main.py 라우터 등록
|
|
|
|
새 라우터를 추가할 때:
|
|
|
|
```python
|
|
# main.py
|
|
from routers import (
|
|
...
|
|
new_module, # ← 여기에 추가
|
|
)
|
|
|
|
# lifespan 함수 안에 필요한 초기화 추가
|
|
|
|
# 라우터 등록 (알파벳 순 권장)
|
|
app.include_router(new_module.router)
|
|
```
|
|
|
|
---
|
|
|
|
## 5. 데이터베이스 가이드
|
|
|
|
### 5.1 마이그레이션 정책
|
|
|
|
현재 GUARDiA ITSM은 **Alembic 없이** `init_db()`로 테이블을 자동 생성합니다.
|
|
|
|
```python
|
|
# database.py
|
|
async def init_db():
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
```
|
|
|
|
> **주의**: 운영 DB에서는 테이블 삭제/재생성이 일어나지 않습니다. 컬럼 추가/변경 시 수동 ALTER TABLE이 필요합니다.
|
|
|
|
**신규 컬럼 추가 절차**:
|
|
1. `models.py`에 컬럼 추가
|
|
2. 운영 DB에 수동 ALTER: `ALTER TABLE tb_xxx ADD COLUMN new_col TEXT;`
|
|
3. 기본값 업데이트: `UPDATE tb_xxx SET new_col = 'default' WHERE new_col IS NULL;`
|
|
|
|
### 5.2 관계 설정 패턴
|
|
|
|
```python
|
|
# 1:N 관계 (부모→자식)
|
|
class SRRequest(Base):
|
|
__tablename__ = "tb_sr_request"
|
|
# ...
|
|
work_logs = relationship("WorkLog", back_populates="sr", lazy="select")
|
|
|
|
class WorkLog(Base):
|
|
__tablename__ = "tb_work_log"
|
|
sr_id = Column(Integer, ForeignKey("tb_sr_request.id"), nullable=False)
|
|
sr = relationship("SRRequest", back_populates="work_logs")
|
|
|
|
# 쿼리 시 eagerly load (N+1 문제 방지)
|
|
from sqlalchemy.orm import selectinload
|
|
query = select(SRRequest).options(selectinload(SRRequest.work_logs))
|
|
```
|
|
|
|
### 5.3 enum 정의 패턴
|
|
|
|
```python
|
|
import enum
|
|
|
|
class SRStatus(str, enum.Enum):
|
|
OPEN = "OPEN"
|
|
IN_PROGRESS = "IN_PROGRESS"
|
|
RESOLVED = "RESOLVED"
|
|
CLOSED = "CLOSED"
|
|
|
|
# 모델에서 사용
|
|
class SRRequest(Base):
|
|
status = Column(Enum(SRStatus), default=SRStatus.OPEN, nullable=False)
|
|
```
|
|
|
|
### 5.4 트랜잭션 처리
|
|
|
|
```python
|
|
# 여러 테이블 수정 시 트랜잭션 보장
|
|
async def complex_operation(db: AsyncSession):
|
|
try:
|
|
obj1 = Model1(...)
|
|
obj2 = Model2(...)
|
|
db.add(obj1)
|
|
db.add(obj2)
|
|
await db.commit() # 모두 성공 시 커밋
|
|
except Exception:
|
|
await db.rollback() # 하나라도 실패 시 롤백
|
|
raise
|
|
```
|
|
|
|
---
|
|
|
|
## 6. 보안 필수 규칙
|
|
|
|
> **이 규칙들은 절대 예외가 없습니다. 위반 시 코드 리뷰에서 반려됩니다.**
|
|
|
|
### 6.1 서버 자격증명 보호
|
|
|
|
```python
|
|
# ✅ 서버 조회 응답에서 민감 필드 제외
|
|
class ServerOut(BaseModel):
|
|
id: int
|
|
hostname: str
|
|
server_role: str
|
|
# ip_addr, ssh_user, os_pw_enc 절대 포함 금지!
|
|
|
|
# ❌ 절대 하지 말 것
|
|
class ServerOut(BaseModel):
|
|
ip_addr: str # ❌ IP 노출 금지
|
|
ssh_user: str # ❌ SSH 계정 노출 금지
|
|
os_pw_enc: str # ❌ 암호화된 비밀번호도 노출 금지
|
|
```
|
|
|
|
### 6.2 SSH 명령 안전성 검증
|
|
|
|
```python
|
|
_DANGEROUS_PATTERNS = [
|
|
"rm -rf /", "rm -rf /*", "mkfs", "dd if=",
|
|
"shutdown", "reboot", "halt", "poweroff",
|
|
":(){:|:&};:", ">()", "chmod 777 /",
|
|
"chown -R root /", "> /dev/sda",
|
|
]
|
|
|
|
def _validate_command(cmd: str) -> None:
|
|
for pattern in _DANGEROUS_PATTERNS:
|
|
if pattern in cmd:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"안전하지 않은 명령어 패턴이 포함되어 있습니다: {pattern}"
|
|
)
|
|
```
|
|
|
|
### 6.3 파일 경로 순회 방지
|
|
|
|
```python
|
|
from pathlib import Path
|
|
|
|
UPLOAD_ROOT = Path(settings.UPLOAD_ROOT).resolve()
|
|
|
|
def _safe_path(filename: str, subdir: str) -> Path:
|
|
target = (UPLOAD_ROOT / subdir / filename).resolve()
|
|
if not str(target).startswith(str(UPLOAD_ROOT)):
|
|
raise HTTPException(status_code=400, detail="잘못된 파일 경로입니다.")
|
|
return target
|
|
```
|
|
|
|
### 6.4 AES-256-GCM 암호화 사용
|
|
|
|
```python
|
|
# utils/crypto.py
|
|
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
|
|
import os, base64
|
|
|
|
def encrypt(plaintext: str, key_hex: str) -> str:
|
|
key = bytes.fromhex(key_hex)
|
|
nonce = os.urandom(12)
|
|
aesgcm = AESGCM(key)
|
|
ct = aesgcm.encrypt(nonce, plaintext.encode(), None)
|
|
return base64.b64encode(nonce + ct).decode()
|
|
|
|
def decrypt(ciphertext: str, key_hex: str) -> str:
|
|
key = bytes.fromhex(key_hex)
|
|
data = base64.b64decode(ciphertext)
|
|
nonce, ct = data[:12], data[12:]
|
|
aesgcm = AESGCM(key)
|
|
return aesgcm.decrypt(nonce, ct, None).decode()
|
|
```
|
|
|
|
### 6.5 Messenger 발송 메시지 보안
|
|
|
|
```python
|
|
# ✅ 올바른 알림 메시지 (서버 정보 제외)
|
|
message = f"[장애] {incident.title}\n등급: {incident.priority}\nID: {incident.incident_id}"
|
|
|
|
# ❌ 절대 금지 (서버 정보 포함)
|
|
message = f"서버 {server.ip_addr}에서 장애 발생" # ❌ IP 노출
|
|
message = f"SSH 계정: {server.ssh_user}" # ❌ 계정 노출
|
|
```
|
|
|
|
### 6.6 JWT 토큰 처리
|
|
|
|
```python
|
|
# 토큰 생성 시 만료 시간 필수
|
|
def create_access_token(data: dict) -> str:
|
|
expire = datetime.utcnow() + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
|
|
return jwt.encode({**data, "exp": expire}, settings.SECRET_KEY, algorithm="HS256")
|
|
|
|
# 의존성에서 자동 검증
|
|
async def get_current_user(token: str = Depends(oauth2_scheme), db: AsyncSession = Depends(get_db)):
|
|
try:
|
|
payload = jwt.decode(token, settings.SECRET_KEY, algorithms=["HS256"])
|
|
except JWTError:
|
|
raise HTTPException(status_code=401, detail="유효하지 않은 토큰입니다.")
|
|
```
|
|
|
|
### 6.7 root SSH 접속 금지
|
|
|
|
```python
|
|
# 서버 등록/수정 시 검증
|
|
if body.ssh_user == "root":
|
|
raise HTTPException(status_code=400, detail="root SSH 직접 접속은 허용되지 않습니다.")
|
|
```
|
|
|
|
### 6.8 라이선스 제한 강제
|
|
|
|
에디션별 자원 한도(기관·사용자·서버 수)를 초과하는 생성 API에는 반드시 Dependency를 추가해야 한다.
|
|
|
|
```python
|
|
from middleware.license_guard import (
|
|
check_institution_limit,
|
|
check_server_limit,
|
|
check_user_limit,
|
|
require_feature,
|
|
)
|
|
|
|
# 기관 등록 엔드포인트 예시
|
|
@router.post("/", response_model=InstitutionOut, status_code=201)
|
|
async def create_institution(
|
|
payload: InstitutionCreate,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
_: None = Depends(check_institution_limit), # 에디션 한도 초과 시 HTTP 403
|
|
):
|
|
...
|
|
|
|
# 서버 등록 엔드포인트 예시
|
|
@router.post("/servers", response_model=ServerOut, status_code=201)
|
|
async def create_server(
|
|
payload: dict,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
_: None = Depends(check_server_limit), # 에디션 한도 초과 시 HTTP 403
|
|
):
|
|
...
|
|
|
|
# 특정 기능이 에디션에 포함되는지 체크 (STANDARD 이상 필요한 LDAP 예시)
|
|
@router.post("/ldap/sync")
|
|
async def ldap_sync(
|
|
_: None = Depends(require_feature("LDAP")),
|
|
):
|
|
...
|
|
```
|
|
|
|
에디션별 기본 한도:
|
|
| 에디션 | 기관 | 사용자 | 서버 |
|
|
|--------|------|-------|------|
|
|
| COMMUNITY | 1 | 10 | 20 |
|
|
| STANDARD | 50 | 200 | 500 |
|
|
| ENTERPRISE | 무제한 | 무제한 | 무제한 |
|
|
|
|
---
|
|
|
|
## 7. 테스트 작성 가이드
|
|
|
|
### 7.1 단위 테스트 기본 구조
|
|
|
|
```python
|
|
import pytest
|
|
import pytest_asyncio
|
|
|
|
class TestSslManager:
|
|
@pytest.mark.asyncio
|
|
async def test_register_domain_success(self, client, auth_headers):
|
|
# Arrange
|
|
payload = {"domain": "example.com", "port": 443, "server_id": 1}
|
|
|
|
# Act
|
|
r = await client.post("/ssl-manager/domains", json=payload, headers=auth_headers)
|
|
|
|
# Assert
|
|
assert r.status_code == 201
|
|
data = r.json()
|
|
assert data["domain"] == "example.com"
|
|
assert "id" in data
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_register_domain_duplicate(self, client, auth_headers):
|
|
payload = {"domain": "example.com", "port": 443, "server_id": 1}
|
|
await client.post("/ssl-manager/domains", json=payload, headers=auth_headers)
|
|
|
|
r = await client.post("/ssl-manager/domains", json=payload, headers=auth_headers)
|
|
assert r.status_code == 409
|
|
```
|
|
|
|
### 7.2 mock 사용 패턴
|
|
|
|
```python
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_p1_incident_triggers_notification(client, auth_headers):
|
|
with patch("routers.incidents._notify_incident", new_callable=AsyncMock) as mock_notify:
|
|
r = await client.post("/incidents/", json={
|
|
"title": "DB 서버 다운",
|
|
"priority": "P1",
|
|
"description": "운영 DB 응답 없음",
|
|
}, headers=auth_headers)
|
|
|
|
assert r.status_code == 201
|
|
mock_notify.assert_called_once()
|
|
call_args = mock_notify.call_args
|
|
assert call_args[0][0].priority == "P1" # 첫 번째 인자 확인
|
|
```
|
|
|
|
### 7.3 시간 고정 테스트
|
|
|
|
```python
|
|
from freezegun import freeze_time
|
|
|
|
@pytest.mark.asyncio
|
|
@freeze_time("2026-01-15 00:00:00")
|
|
async def test_pm_schedule_monthly_next_date(client, auth_headers):
|
|
r = await client.post("/pm/schedules/", json={
|
|
"template_id": 1,
|
|
"frequency": "MONTHLY",
|
|
"day_of_month": 15,
|
|
"server_id": 1,
|
|
}, headers=auth_headers)
|
|
assert r.status_code == 201
|
|
# 1월 15일 기준 → 다음 실행은 2월 15일
|
|
assert "2026-02-15" in r.json()["next_scheduled"]
|
|
```
|
|
|
|
---
|
|
|
|
## 8. Messenger Bot 개발 가이드
|
|
|
|
### 8.1 GUARDiA_ITSM 봇 명령 추가
|
|
|
|
명령어 추가는 `C:\GUARDiA\messenger\core\itsm_bot.py` 에서 수행합니다.
|
|
|
|
```python
|
|
# 새 명령어 핸들러 추가 패턴
|
|
COMMAND_HANDLERS = {
|
|
"/itsm sr list": handle_sr_list,
|
|
"/itsm sr create": handle_sr_create,
|
|
"/itsm help": handle_help,
|
|
"/itsm incident p1": handle_p1_alert, # 새 명령어 추가
|
|
}
|
|
|
|
async def handle_p1_alert(args: list[str], user_id: int) -> str:
|
|
"""P1 장애 목록 조회"""
|
|
async with httpx.AsyncClient() as client:
|
|
r = await client.get(
|
|
f"{ITSM_BASE_URL}/incidents/?priority=P1&status=OPEN",
|
|
headers={"Authorization": f"Bearer {ITSM_BOT_TOKEN}"},
|
|
)
|
|
if r.status_code != 200:
|
|
return "장애 목록 조회 실패"
|
|
incidents = r.json()["items"]
|
|
if not incidents:
|
|
return "현재 P1 장애 없음"
|
|
lines = [f"[{i['incident_id']}] {i['title']}" for i in incidents[:5]]
|
|
return "P1 장애 목록:\n" + "\n".join(lines)
|
|
```
|
|
|
|
### 8.2 ITSM → Messenger 알림 발송
|
|
|
|
```python
|
|
# routers/incidents.py 내 알림 함수
|
|
async def _notify_incident(incident: IncidentModel) -> None:
|
|
"""장애 발생 시 Messenger 채널 알림"""
|
|
try:
|
|
channel_id = settings.INCIDENT_CHANNEL_ID
|
|
# 보안: 서버 IP/PW 미포함
|
|
msg = (
|
|
f"🚨 **{incident.priority} 장애 발생**\n"
|
|
f"ID: {incident.incident_id}\n"
|
|
f"제목: {incident.title}\n"
|
|
f"등록: {incident.created_at.strftime('%Y-%m-%d %H:%M')}"
|
|
)
|
|
async with httpx.AsyncClient(timeout=5) as client:
|
|
await client.post(
|
|
f"{settings.MESSENGER_BASE_URL}/api/messages",
|
|
json={"channel_id": channel_id, "content": msg},
|
|
headers={"Authorization": f"Bearer {settings.MESSENGER_BOT_TOKEN}"},
|
|
)
|
|
except Exception:
|
|
# 알림 실패가 비즈니스 로직을 방해하면 안 됨
|
|
pass
|
|
```
|
|
|
|
---
|
|
|
|
## 9. 스케줄러 확장 가이드
|
|
|
|
### 9.1 새 스케줄 작업 추가
|
|
|
|
`core/scheduler.py`에 새 작업을 추가합니다:
|
|
|
|
```python
|
|
from apscheduler.triggers.cron import CronTrigger
|
|
|
|
async def _new_scheduled_task() -> None:
|
|
"""매주 월요일 오전 9시 실행 예시"""
|
|
from database import SessionLocal
|
|
async with SessionLocal() as db:
|
|
# 작업 수행
|
|
pass
|
|
|
|
def start_scheduler():
|
|
# 기존 작업들 ...
|
|
_scheduler.add_job(
|
|
_new_scheduled_task,
|
|
CronTrigger(day_of_week="mon", hour=9, minute=0),
|
|
id="new_task",
|
|
replace_existing=True,
|
|
)
|
|
_scheduler.start()
|
|
```
|
|
|
|
### 9.2 스케줄러 테스트
|
|
|
|
스케줄러는 직접 함수 호출로 테스트합니다:
|
|
|
|
```python
|
|
@pytest.mark.asyncio
|
|
async def test_ssl_expiry_scan(db_session):
|
|
from core.scheduler import _scan_ssl_expiry
|
|
# 테스트 데이터 준비
|
|
# ...
|
|
await _scan_ssl_expiry() # 직접 호출
|
|
# DB 상태 확인
|
|
# ...
|
|
```
|
|
|
|
---
|
|
|
|
## 10. 배포 및 CI/CD
|
|
|
|
### 10.1 Jenkins 파이프라인 단계
|
|
|
|
```groovy
|
|
pipeline {
|
|
stages {
|
|
stage('Test') {
|
|
steps {
|
|
sh 'pytest tests/unit/ --cov=. --cov-report=xml'
|
|
}
|
|
}
|
|
stage('Quality Gate') {
|
|
steps {
|
|
// SonarQube 분석
|
|
sh 'sonar-scanner -Dsonar.projectKey=guardia-itsm'
|
|
}
|
|
}
|
|
stage('Deploy') {
|
|
when { branch 'main' }
|
|
steps {
|
|
sh './scripts/deploy.sh production'
|
|
}
|
|
}
|
|
}
|
|
}
|
|
```
|
|
|
|
### 10.2 운영 서버 배포 체크리스트
|
|
|
|
```
|
|
배포 전:
|
|
□ 모든 단위 테스트 통과
|
|
□ 커버리지 80% 이상
|
|
□ SonarQube 품질 게이트 통과
|
|
□ .env 운영 설정 확인 (SECRET_KEY, ENCRYPTION_KEY)
|
|
□ DB 백업 완료
|
|
|
|
배포 중:
|
|
□ 서비스 점검 공지
|
|
□ 현재 버전 백업
|
|
□ 새 버전 배포
|
|
□ DB 마이그레이션 (필요 시)
|
|
□ 서비스 재시작
|
|
|
|
배포 후:
|
|
□ /docs 페이지 접근 확인
|
|
□ 로그인 테스트
|
|
□ 주요 API 스모크 테스트
|
|
□ 스케줄러 동작 확인
|
|
```
|
|
|
|
---
|
|
|
|
## 11. 자주 하는 실수 및 주의사항
|
|
|
|
### 11.1 비동기 함수 실수
|
|
|
|
```python
|
|
# ❌ 실수: async 함수에서 동기 DB 호출
|
|
async def bad_function(db: AsyncSession):
|
|
result = db.execute(select(Item)) # ❌ await 누락
|
|
|
|
# ✅ 올바른 방법
|
|
async def good_function(db: AsyncSession):
|
|
result = await db.execute(select(Item)) # ✅
|
|
```
|
|
|
|
### 11.2 DB 세션 누수
|
|
|
|
```python
|
|
# ❌ 실수: 백그라운드 태스크에서 request 세션 사용
|
|
@router.post("/")
|
|
async def create(background_tasks: BackgroundTasks, db: AsyncSession = Depends(get_db)):
|
|
background_tasks.add_task(some_task, db) # ❌ 세션이 request 후 닫힘
|
|
|
|
# ✅ 올바른 방법: 백그라운드에서 새 세션 생성
|
|
async def some_task():
|
|
from database import SessionLocal
|
|
async with SessionLocal() as db: # ✅ 새 세션
|
|
pass
|
|
```
|
|
|
|
### 11.3 N+1 쿼리 문제
|
|
|
|
```python
|
|
# ❌ N+1 쿼리 발생
|
|
srs = (await db.execute(select(SRRequest))).scalars().all()
|
|
for sr in srs:
|
|
print(sr.work_logs) # 각 SR마다 추가 쿼리 발생!
|
|
|
|
# ✅ selectinload 사용
|
|
srs = (await db.execute(
|
|
select(SRRequest).options(selectinload(SRRequest.work_logs))
|
|
)).scalars().all()
|
|
```
|
|
|
|
### 11.4 환경 변수 검증
|
|
|
|
```python
|
|
# ✅ 앱 시작 시 필수 환경 변수 검증
|
|
class Settings(BaseSettings):
|
|
SECRET_KEY: str
|
|
ENCRYPTION_KEY: str
|
|
|
|
@validator("ENCRYPTION_KEY")
|
|
def encryption_key_must_be_64_hex(cls, v):
|
|
if len(v) != 64:
|
|
raise ValueError("ENCRYPTION_KEY는 64자리 hex 문자열이어야 합니다.")
|
|
return v
|
|
```
|
|
|
|
### 11.5 외부 API 완전 금지
|
|
|
|
```python
|
|
# ❌ 절대 금지: 외부 LLM/AI API 호출
|
|
import openai
|
|
client = openai.OpenAI(api_key="...") # ❌ 외부 API 사용 금지
|
|
|
|
# ✅ 내부 sLLM만 허용
|
|
async with httpx.AsyncClient() as client:
|
|
r = await client.post(f"{settings.SLLM_BASE_URL}/v1/chat/completions", ...)
|
|
```
|
|
|
|
---
|
|
|
|
*본 지침서를 준수하지 않은 코드는 코드 리뷰에서 반려될 수 있습니다.*
|
|
*보안 관련 규칙(6장)은 특히 엄격히 적용됩니다.*
|