refactor: 101.79.17.164 → zioinfo.co.kr 전체 도메인 변환 + Manager UI 배포

- 37개 파일 IP → zioinfo.co.kr 치환 (소스/매뉴얼/설정/하네스)
- Manager DrConsole/NetworkConsole/CsapConsole 빌드 + /var/www/manager/ 배포
- 테스트: Manager HTTP 200, ITSM 신규 API 7개 전체 200

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Jenkinsfile

@@ -0,0 +1,91 @@
+pipeline {
+    agent any
+
+    environment {
+        APP_DIR   = '/opt/guardia/app'
+        VENV      = '/opt/guardia/venv'
+        SERVICE   = 'guardia'
+        GITEA_URL = 'http://localhost:3000/zio/guardia-itsm.git'
+    }
+
+    options {
+        buildDiscarder(logRotator(numToKeepStr: '5'))
+        timeout(time: 15, unit: 'MINUTES')
+    }
+
+    stages {
+        stage('Checkout') {
+            steps {
+                echo "체크아웃: ${env.GIT_BRANCH ?: 'main'}"
+                checkout scm
+            }
+        }
+
+        stage('Python Lint') {
+            steps {
+                sh '''
+                    echo "=== Python 문법 검사 ==="
+                    python3 -m py_compile main.py models.py database.py
+                    find routers/ -name "*.py" -exec python3 -m py_compile {} \\;
+                    find core/ -name "*.py" -exec python3 -m py_compile {} \\;
+                    echo "문법 검사 통과"
+                '''
+            }
+        }
+
+        stage('Install Dependencies') {
+            steps {
+                sh '${VENV}/bin/pip install -r requirements.txt -q && echo "의존성 OK"'
+            }
+        }
+
+        stage('Health Check') {
+            steps {
+                sh '''
+                    if systemctl is-active ${SERVICE} >/dev/null 2>&1; then
+                        HTTP=$(curl -s -o /dev/null -w "%{http_code}" http://localhost:8001/docs)
+                        echo "현재 서비스 HTTP: $HTTP"
+                    else
+                        echo "서비스 미실행"
+                    fi
+                '''
+            }
+        }
+
+        stage('Deploy') {
+            when { branch 'main' }
+            steps {
+                sh '''
+                    echo "=== GUARDiA ITSM 배포 ==="
+                    # 백업
+                    BACKUP=/opt/guardia/backups/$(date +%Y%m%d_%H%M%S)
+                    mkdir -p $BACKUP
+                    cp -r ${APP_DIR}/*.py ${APP_DIR}/routers ${APP_DIR}/core $BACKUP/ 2>/dev/null || true
+
+                    # 파일 복사
+                    rsync -av --exclude="__pycache__" --exclude="test_*.py" \\
+                        --exclude="*.db" --exclude=".git" \\
+                        ./ ${APP_DIR}/
+
+                    # 패키지 업데이트
+                    ${VENV}/bin/pip install -r requirements.txt -q
+
+                    # 서비스 재시작
+                    systemctl restart ${SERVICE}
+                    sleep 5
+
+                    # 헬스체크
+                    HTTP=$(curl -s -o /dev/null -w "%{http_code}" http://localhost:8001/docs)
+                    echo "배포 후 HTTP: $HTTP"
+                    [ "$HTTP" = "200" ] && echo "배포 성공" || (echo "배포 실패"; exit 1)
+                '''
+            }
+        }
+    }
+
+    post {
+        success { echo "✅ GUARDiA 배포 성공: ${currentBuild.displayName}" }
+        failure { echo "❌ GUARDiA 배포 실패: ${currentBuild.displayName}" }
+        always  { cleanWs(cleanWhenNotBuilt: false, cleanWhenSuccess: false) }
+    }
+}

core/external_security.py

@@ -0,0 +1,94 @@
+"""
+개방망 외부 API 보안 — API Key 발급·검증·감사
+- API Key: sha256 해시 저장, 평문은 발급 시 1회만 노출
+- 권한 스코프: read | write | admin | webhook
+- IP 화이트리스트 (선택)
+- 요청별 감사 로깅
+"""
+import hashlib
+import secrets
+import time
+from datetime import datetime, timezone
+from typing import Optional
+
+from fastapi import Depends, Header, HTTPException, Request, status
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from database import get_db
+from models import APIKey, AuditLog
+
+
+# ── API Key 생성 ──────────────────────────────────────────────────────────────
+
+def generate_api_key() -> tuple[str, str]:
+    """평문 키와 SHA-256 해시를 반환. 평문은 1회만 노출."""
+    plain = "grd_" + secrets.token_urlsafe(32)
+    hashed = hashlib.sha256(plain.encode()).hexdigest()
+    return plain, hashed
+
+
+def hash_key(plain: str) -> str:
+    return hashlib.sha256(plain.encode()).hexdigest()
+
+
+# ── API Key 검증 Dependency ───────────────────────────────────────────────────
+
+async def verify_api_key(
+    request: Request,
+    x_api_key: Optional[str] = Header(None, alias="X-API-Key"),
+    db: AsyncSession = Depends(get_db),
+) -> "APIKey":
+    """
+    외부 API 호출용 API Key 인증.
+    헤더: X-API-Key: grd_xxxxx
+    """
+    if not x_api_key:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="X-API-Key 헤더가 필요합니다.",
+            headers={"WWW-Authenticate": "ApiKey"},
+        )
+
+    key_hash = hash_key(x_api_key)
+    row = await db.execute(
+        select(APIKey).where(APIKey.key_hash == key_hash, APIKey.is_active == True)
+    )
+    api_key = row.scalar_one_or_none()
+
+    if not api_key:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="유효하지 않은 API Key입니다.")
+
+    # 만료 확인
+    if api_key.expires_at and api_key.expires_at < datetime.now(timezone.utc).replace(tzinfo=None):
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="만료된 API Key입니다.")
+
+    # IP 화이트리스트 확인
+    if api_key.allowed_ips:
+        client_ip = request.client.host
+        allowed = [ip.strip() for ip in api_key.allowed_ips.split(",")]
+        if client_ip not in allowed:
+            raise HTTPException(
+                status_code=status.HTTP_403_FORBIDDEN,
+                detail=f"허용되지 않은 IP입니다.",
+            )
+
+    # 사용 횟수 갱신
+    api_key.use_count = (api_key.use_count or 0) + 1
+    api_key.last_used_at = datetime.utcnow()
+    await db.commit()
+
+    return api_key
+
+
+def require_scope(scope: str):
+    """특정 스코프(read/write/admin/webhook)를 요구하는 Dependency 팩토리."""
+    async def _check(api_key: "APIKey" = Depends(verify_api_key)):
+        scopes = (api_key.scopes or "").split(",")
+        if scope not in scopes and "admin" not in scopes:
+            raise HTTPException(
+                status_code=status.HTTP_403_FORBIDDEN,
+                detail=f"'{scope}' 권한이 필요합니다.",
+            )
+        return api_key
+    return _check

deploy/guardia_deploy.sh

@@ -0,0 +1,48 @@
+#!/bin/bash
+# GUARDiA ITSM 자동 배포 스크립트 (Deploy Webhook / CI 트리거 시 실행)
+# 경로: /opt/guardia/app/deploy/guardia_deploy.sh
+
+set -euo pipefail
+LOG=/opt/guardia/logs/deploy.log
+APP=/opt/guardia/app
+REPO=http://zio:Zio%40Admin2026%21@localhost:3000/zio/guardia-itsm.git
+SRC=/opt/guardia/src
+
+echo "[$(date '+%Y-%m-%d %H:%M:%S')] === GUARDiA 배포 시작 ===" >> $LOG
+
+# 1. 소스 갱신
+if [ ! -d "$SRC/.git" ]; then
+    echo "[$(date)] git clone..." >> $LOG
+    git clone $REPO $SRC
+else
+    echo "[$(date)] git pull..." >> $LOG
+    cd $SRC && git pull origin main
+fi
+echo "[$(date)] 소스 갱신 완료" >> $LOG
+
+# 2. 새 파일 복사 (test_, __pycache__ 제외)
+rsync -av --exclude="__pycache__" --exclude="test_*.py" \
+    --exclude="*.db" --exclude="*.bak" --exclude=".git" \
+    $SRC/ $APP/ >> $LOG 2>&1
+echo "[$(date)] 파일 복사 완료" >> $LOG
+
+# 3. 패키지 업데이트
+if [ -f "$SRC/requirements.txt" ]; then
+    /opt/guardia/venv/bin/pip install -r $SRC/requirements.txt -q >> $LOG 2>&1
+    echo "[$(date)] 패키지 업데이트 완료" >> $LOG
+fi
+
+# 4. 서비스 재시작
+systemctl restart guardia
+sleep 5
+
+# 5. 헬스체크
+HTTP=$(curl -s -o /dev/null -w "%{http_code}" http://localhost:8001/docs 2>/dev/null)
+if [ "$HTTP" = "200" ]; then
+    echo "[$(date)] 배포 성공 (HTTP $HTTP)" >> $LOG
+else
+    echo "[$(date)] 배포 실패 (HTTP $HTTP)" >> $LOG
+    exit 1
+fi
+
+echo "[$(date)] === GUARDiA 배포 완료 ===" >> $LOG

routers/export_import.py

@@ -0,0 +1,321 @@
+"""
+폐쇄망 ↔ 개방망 데이터 Export / Import 인터페이스
+
+Export (폐쇄망 → 파일):
+  GET /api/export/bundle          — 전체 데이터 번들 (JSON ZIP)
+  GET /api/export/sr              — SR 목록 JSON
+  GET /api/export/cmdb            — CMDB 서버 자산 JSON
+  GET /api/export/kb              — 지식베이스 JSON
+  GET /api/export/audit           — 감사 로그 JSON
+
+Import (파일 → 개방망 GUARDiA):
+  POST /api/import/bundle         — 번들 ZIP 업로드 및 병합
+  POST /api/import/sr             — SR JSON 업로드
+  POST /api/import/cmdb           — CMDB JSON 업로드
+
+보안:
+  - Admin JWT 인증 필수
+  - 민감 필드(ip_addr, os_pw_enc, ssh_user) Export 시 마스킹
+  - 번들에 HMAC-SHA256 서명 포함
+"""
+import hashlib, hmac, io, json, os, zipfile
+from datetime import datetime
+from typing import Optional
+
+from fastapi import APIRouter, Depends, File, HTTPException, Query, UploadFile
+from fastapi.responses import StreamingResponse
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from core.auth import get_current_user, require_admin_role
+from database import get_db
+from models import (
+    SRRequest, Institution, Server, AuditLog, User,
+)
+
+router = APIRouter(prefix="/api/export-import", tags=["Export/Import (폐쇄망 연동)"])
+
+BUNDLE_SECRET = os.environ.get("GUARDIA_BUNDLE_SECRET", "guardia-bundle-hmac-2026")
+SENSITIVE_FIELDS = {"ip_addr", "os_pw_enc", "ssh_user", "ssh_pw"}
+
+
+# ── 유틸 ──────────────────────────────────────────────────────────────────────
+
+def _mask_row(row: dict) -> dict:
+    return {k: ("****" if k in SENSITIVE_FIELDS else v) for k, v in row.items()}
+
+def _sign_bundle(data: bytes) -> str:
+    return hmac.new(BUNDLE_SECRET.encode(), data, hashlib.sha256).hexdigest()
+
+def _row_to_dict(obj) -> dict:
+    from datetime import date
+    result = {}
+    for col in obj.__table__.columns:
+        val = getattr(obj, col.name)
+        if isinstance(val, datetime):
+            val = val.isoformat()
+        elif isinstance(val, date):
+            val = val.isoformat()
+        elif val is not None and not isinstance(val, (str, int, float, bool, dict, list)):
+            val = str(val)
+        result[col.name] = val
+    return _mask_row(result)
+
+
+# ── Export 엔드포인트 ─────────────────────────────────────────────────────────
+
+@router.get("/export/sr", summary="SR 목록 Export (JSON)")
+async def export_sr(
+    limit: int = Query(default=1000, le=5000),
+    since: Optional[str] = Query(default=None, description="ISO 날짜, 예: 2026-01-01"),
+    db: AsyncSession = Depends(get_db),
+    _user=Depends(require_admin_role),
+):
+    q = select(SRRequest).order_by(SRRequest.created_at.desc()).limit(limit)
+    if since:
+        try:
+            dt = datetime.fromisoformat(since).replace(tzinfo=None)
+            q = q.where(SRRequest.created_at >= dt)
+        except ValueError:
+            raise HTTPException(status_code=400, detail="since 날짜 형식 오류 (ISO 8601)")
+    rows = (await db.execute(q)).scalars().all()
+    data = [_row_to_dict(r) for r in rows]
+    return {
+        "exported_at": datetime.utcnow().isoformat(),
+        "count": len(data),
+        "type": "sr",
+        "data": data,
+    }
+
+
+@router.get("/export/cmdb", summary="CMDB 서버 자산 Export (JSON)")
+async def export_cmdb(
+    db: AsyncSession = Depends(get_db),
+    _user=Depends(require_admin_role),
+):
+    rows = (await db.execute(select(Server))).scalars().all()
+    data = [_row_to_dict(r) for r in rows]
+    return {
+        "exported_at": datetime.utcnow().isoformat(),
+        "count": len(data),
+        "type": "cmdb",
+        "data": data,
+    }
+
+
+@router.get("/export/institutions", summary="기관 목록 Export (JSON)")
+async def export_institutions(
+    db: AsyncSession = Depends(get_db),
+    _user=Depends(require_admin_role),
+):
+    rows = (await db.execute(select(Institution))).scalars().all()
+    data = [_row_to_dict(r) for r in rows]
+    return {
+        "exported_at": datetime.utcnow().isoformat(),
+        "count": len(data),
+        "type": "institutions",
+        "data": data,
+    }
+
+
+@router.get("/export/audit", summary="감사 로그 Export (JSON)")
+async def export_audit(
+    limit: int = Query(default=500, le=2000),
+    db: AsyncSession = Depends(get_db),
+    _user=Depends(require_admin_role),
+):
+    rows = (await db.execute(
+        select(AuditLog).order_by(AuditLog.created_at.desc()).limit(limit)
+    )).scalars().all()
+    data = [_row_to_dict(r) for r in rows]
+    return {
+        "exported_at": datetime.utcnow().isoformat(),
+        "count": len(data),
+        "type": "audit",
+        "data": data,
+    }
+
+
+@router.get("/export/bundle", summary="전체 번들 Export (ZIP 다운로드)")
+async def export_bundle(
+    db: AsyncSession = Depends(get_db),
+    user=Depends(require_admin_role),
+):
+    """
+    SR, CMDB, 기관, 감사로그를 하나의 ZIP으로 패키징합니다.
+    - 민감 필드(IP, 비밀번호)는 마스킹
+    - 번들에 HMAC-SHA256 서명 포함
+    - Content-Type: application/zip
+    """
+    bundle: dict[str, dict] = {}
+
+    for name, q in [
+        ("sr",           select(SRRequest).order_by(SRRequest.created_at.desc()).limit(2000)),
+        ("cmdb",         select(Server)),
+        ("institutions", select(Institution)),
+        ("audit",        select(AuditLog).order_by(AuditLog.created_at.desc()).limit(1000)),
+    ]:
+        rows = (await db.execute(q)).scalars().all()
+        bundle[name] = {
+            "exported_at": datetime.utcnow().isoformat(),
+            "count": len(rows),
+            "type": name,
+            "data": [_row_to_dict(r) for r in rows],
+        }
+
+    manifest = {
+        "version": "1.0",
+        "source": "GUARDiA ITSM (폐쇄망)",
+        "exported_at": datetime.utcnow().isoformat(),
+        "exported_by": user.username,
+        "contents": list(bundle.keys()),
+        "signature": "",
+    }
+    bundle_json = json.dumps(bundle, ensure_ascii=False, indent=2).encode()
+    manifest["signature"] = _sign_bundle(bundle_json)
+
+    # ZIP 생성
+    buf = io.BytesIO()
+    with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf:
+        zf.writestr("manifest.json", json.dumps(manifest, ensure_ascii=False, indent=2))
+        zf.writestr("bundle.json",   bundle_json)
+        for name, section in bundle.items():
+            zf.writestr(f"{name}.json", json.dumps(section, ensure_ascii=False, indent=2))
+    buf.seek(0)
+
+    filename = f"guardia_export_{datetime.utcnow().strftime('%Y%m%d_%H%M%S')}.zip"
+    return StreamingResponse(
+        buf,
+        media_type="application/zip",
+        headers={"Content-Disposition": f"attachment; filename={filename}"},
+    )
+
+
+# ── Import 엔드포인트 ─────────────────────────────────────────────────────────
+
+@router.post("/import/bundle", summary="번들 ZIP Import (폐쇄망 데이터 업로드)")
+async def import_bundle(
+    file: UploadFile = File(..., description="export_bundle에서 생성된 ZIP 파일"),
+    dry_run: bool = Query(default=True, description="True: 검증만 (실제 저장 안 함)"),
+    db: AsyncSession = Depends(get_db),
+    _user=Depends(require_admin_role),
+):
+    """
+    폐쇄망에서 Export한 번들 ZIP을 업로드합니다.
+    - HMAC 서명 검증
+    - dry_run=True (기본): 데이터 검증만 수행 (DB 저장 없음)
+    - dry_run=False: 실제 병합 (중복 sr_id는 SKIP)
+    """
+    if not file.filename.endswith(".zip"):
+        raise HTTPException(status_code=400, detail="ZIP 파일만 업로드 가능합니다.")
+
+    content = await file.read()
+    if len(content) > 50 * 1024 * 1024:  # 50MB 제한
+        raise HTTPException(status_code=413, detail="파일 크기 50MB 초과")
+
+    try:
+        buf = io.BytesIO(content)
+        with zipfile.ZipFile(buf, "r") as zf:
+            names = zf.namelist()
+            if "manifest.json" not in names or "bundle.json" not in names:
+                raise HTTPException(status_code=400, detail="유효하지 않은 번들 파일입니다.")
+
+            manifest = json.loads(zf.read("manifest.json"))
+            bundle_json = zf.read("bundle.json")
+            bundle = json.loads(bundle_json)
+    except (zipfile.BadZipFile, json.JSONDecodeError) as e:
+        raise HTTPException(status_code=400, detail=f"파일 파싱 오류: {e}")
+
+    # HMAC 검증
+    expected_sig = _sign_bundle(bundle_json)
+    if manifest.get("signature") != expected_sig:
+        raise HTTPException(status_code=400, detail="번들 서명 검증 실패 (위변조 의심)")
+
+    stats = {
+        "manifest": {
+            "exported_at": manifest.get("exported_at"),
+            "source":      manifest.get("source"),
+            "exported_by": manifest.get("exported_by"),
+        },
+        "sections": {},
+    }
+
+    for section_name, section in bundle.items():
+        count = section.get("count", 0)
+        stats["sections"][section_name] = {
+            "count": count, "status": "dry_run" if dry_run else "imported",
+        }
+
+    if not dry_run:
+        # SR Import (중복 sr_id SKIP)
+        sr_section = bundle.get("sr", {})
+        imported = 0
+        for row in sr_section.get("data", []):
+            existing = await db.execute(
+                select(SRRequest).where(SRRequest.sr_id == row.get("sr_id"))
+            )
+            if existing.scalars().first():
+                continue
+            sr = SRRequest(
+                sr_id       = row.get("sr_id", f"IMP-{hashlib.md5(str(row).encode()).hexdigest()[:8].upper()}"),
+                title       = row.get("title", "(imported)"),
+                description = row.get("description", ""),
+                status      = row.get("status", "RECEIVED"),
+                priority    = row.get("priority", "MEDIUM"),
+                sr_type     = row.get("sr_type", "OTHER"),
+                requested_by= row.get("requested_by", "imported"),
+            )
+            db.add(sr); imported += 1
+        await db.commit()
+        stats["sections"]["sr"]["imported"] = imported
+
+    return {
+        "status":  "dry_run" if dry_run else "imported",
+        "message": "검증 완료 (dry_run)" if dry_run else f"Import 완료",
+        **stats,
+    }
+
+
+@router.post("/import/sr", summary="SR JSON Import")
+async def import_sr(
+    file: UploadFile = File(..., description="export/sr에서 내보낸 JSON 파일"),
+    dry_run: bool = Query(default=True),
+    db: AsyncSession = Depends(get_db),
+    _user=Depends(require_admin_role),
+):
+    content = await file.read()
+    try:
+        data = json.loads(content)
+        records = data.get("data", data) if isinstance(data, dict) else data
+    except json.JSONDecodeError as e:
+        raise HTTPException(status_code=400, detail=f"JSON 파싱 오류: {e}")
+
+    if not isinstance(records, list):
+        raise HTTPException(status_code=400, detail="data 배열이 필요합니다.")
+
+    imported = 0
+    if not dry_run:
+        for row in records:
+            existing = await db.execute(
+                select(SRRequest).where(SRRequest.sr_id == row.get("sr_id"))
+            )
+            if existing.scalars().first():
+                continue
+            db.add(SRRequest(
+                sr_id       = row.get("sr_id", f"IMP-{imported:06d}"),
+                title       = row.get("title", "(imported)"),
+                description = row.get("description", ""),
+                status      = row.get("status", "RECEIVED"),
+                priority    = row.get("priority", "MEDIUM"),
+                sr_type     = row.get("sr_type", "OTHER"),
+                requested_by= row.get("requested_by", "imported"),
+            ))
+            imported += 1
+        await db.commit()
+
+    return {
+        "status":   "dry_run" if dry_run else "imported",
+        "total":    len(records),
+        "imported": imported,
+        "message":  f"SR {'검증' if dry_run else '가져오기'} 완료: {len(records)}건",
+    }

routers/external_api.py

@@ -0,0 +1,261 @@
+"""
+개방망 외부 API 라우터
+- /api/external/keys        : API Key CRUD (관리자 JWT 인증)
+- /api/external/health      : 공개 헬스체크
+- /api/external/webhook     : 외부 메신저 웹훅 수신
+- /api/external/sr          : 외부에서 SR 조회·등록 (API Key)
+- /api/external/notify      : 외부 알림 전송 (API Key)
+- /api/external/status      : 시스템 상태 공개 요약
+"""
+import hashlib
+import hmac
+import json
+import os
+from datetime import datetime, timedelta
+from typing import Optional
+
+from fastapi import APIRouter, Body, Depends, Header, HTTPException, Request, status
+from pydantic import BaseModel
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from core.external_security import generate_api_key, hash_key, require_scope, verify_api_key
+from core.auth import get_current_user
+from database import get_db
+from models import APIKey, SRRequest as ServiceRequest, User
+
+router = APIRouter(prefix="/api/external", tags=["External API (개방망)"])
+
+
+# ── 공개 엔드포인트 ───────────────────────────────────────────────────────────
+
+@router.get("/health", summary="헬스체크 (인증 불필요)")
+async def health():
+    return {
+        "status": "ok",
+        "service": "GUARDiA ITSM",
+        "version": "2.0.0",
+        "timestamp": datetime.utcnow().isoformat(),
+    }
+
+
+@router.get("/status", summary="시스템 공개 상태")
+async def public_status(db: AsyncSession = Depends(get_db)):
+    """인증 없이 시스템 가동 상태를 반환합니다."""
+    try:
+        sr_count = (await db.execute(
+            select(ServiceRequest).where(ServiceRequest.status != "CLOSED")
+        )).scalars().all()
+        open_sr = len(sr_count)
+    except Exception:
+        open_sr = -1
+
+    return {
+        "status": "operational",
+        "open_service_requests": open_sr,
+        "api_version": "v2",
+        "docs": "/docs",
+    }
+
+
+# ── API Key 관리 (관리자 전용) ────────────────────────────────────────────────
+
+class APIKeyCreate(BaseModel):
+    name: str
+    scopes: str = "read"          # read, write, admin, webhook (쉼표 구분)
+    allowed_ips: Optional[str] = None   # "1.2.3.4,5.6.7.8" (빈 문자열 = 전체 허용)
+    expires_days: Optional[int] = 365
+
+
+@router.get("/keys", summary="API Key 목록 (관리자)")
+async def list_keys(
+    db: AsyncSession = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    if current_user.role not in ("admin", "pm"):
+        raise HTTPException(status_code=403, detail="관리자 권한이 필요합니다.")
+    rows = (await db.execute(select(APIKey).order_by(APIKey.created_at.desc()))).scalars().all()
+    return [
+        {
+            "id": k.id,
+            "name": k.name,
+            "scopes": k.scopes,
+            "allowed_ips": k.allowed_ips,
+            "is_active": k.is_active,
+            "use_count": k.use_count,
+            "last_used_at": k.last_used_at,
+            "expires_at": k.expires_at,
+            "created_at": k.created_at,
+        }
+        for k in rows
+    ]
+
+
+@router.post("/keys", summary="API Key 발급 (관리자)", status_code=201)
+async def create_key(
+    body: APIKeyCreate,
+    db: AsyncSession = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    if current_user.role not in ("admin", "pm"):
+        raise HTTPException(status_code=403, detail="관리자 권한이 필요합니다.")
+
+    plain, key_hash = generate_api_key()
+    expires_at = None
+    if body.expires_days:
+        expires_at = datetime.utcnow() + timedelta(days=body.expires_days)
+
+    api_key = APIKey(
+        name=body.name,
+        key_hash=key_hash,
+        scopes=body.scopes,
+        allowed_ips=body.allowed_ips or "",
+        expires_at=expires_at,
+        created_by=current_user.username,
+        is_active=True,
+        use_count=0,
+    )
+    db.add(api_key)
+    await db.commit()
+    await db.refresh(api_key)
+
+    return {
+        "id": api_key.id,
+        "name": api_key.name,
+        "api_key": plain,          # ← 단 1회만 노출
+        "scopes": api_key.scopes,
+        "expires_at": api_key.expires_at,
+        "warning": "이 키는 다시 조회할 수 없습니다. 안전한 곳에 저장하세요.",
+    }
+
+
+@router.delete("/keys/{key_id}", summary="API Key 비활성화")
+async def revoke_key(
+    key_id: int,
+    db: AsyncSession = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    if current_user.role not in ("admin", "pm"):
+        raise HTTPException(status_code=403, detail="관리자 권한이 필요합니다.")
+    row = await db.get(APIKey, key_id)
+    if not row:
+        raise HTTPException(status_code=404, detail="키를 찾을 수 없습니다.")
+    row.is_active = False
+    await db.commit()
+    return {"message": f"API Key '{row.name}' 비활성화 완료"}
+
+
+# ── 외부 메신저 웹훅 ─────────────────────────────────────────────────────────
+
+WEBHOOK_SECRET = os.environ.get("GUARDIA_WEBHOOK_SECRET", "guardia-webhook-2026")
+
+
+def _verify_hmac(body: bytes, signature: str) -> bool:
+    expected = hmac.new(WEBHOOK_SECRET.encode(), body, hashlib.sha256).hexdigest()
+    return hmac.compare_digest(expected, signature.removeprefix("sha256="))
+
+
+@router.post("/webhook", summary="외부 메신저 웹훅 수신")
+async def receive_webhook(
+    request: Request,
+    x_guardia_signature: Optional[str] = Header(None, alias="X-GUARDiA-Signature"),
+    x_source: str = Header("unknown", alias="X-Source"),
+    db: AsyncSession = Depends(get_db),
+):
+    """
+    카카오워크/네이버웍스/Slack/Teams 등 외부 메신저에서
+    GUARDiA 명령을 전달받는 엔드포인트.
+
+    Headers:
+      X-GUARDiA-Signature: sha256=<hmac>  (선택, 권장)
+      X-Source: kakaotalk | naverworks | slack | teams | custom
+    """
+    body = await request.body()
+
+    # HMAC 검증 (서명이 있는 경우만 강제)
+    if x_guardia_signature:
+        if not _verify_hmac(body, x_guardia_signature):
+            raise HTTPException(status_code=401, detail="웹훅 서명 검증 실패")
+
+    try:
+        payload = json.loads(body)
+    except Exception:
+        raise HTTPException(status_code=400, detail="JSON 파싱 오류")
+
+    command = payload.get("command") or payload.get("text") or payload.get("message", "")
+    user_id = payload.get("user_id") or payload.get("sender") or "external"
+
+    # AI 자연어 명령 파싱 후 처리
+    try:
+        from core.nlu import parse_command
+        parsed = await parse_command(command, db)
+        action = parsed.get("action", "unknown")
+    except Exception:
+        parsed = {"action": "echo", "raw": command}
+        action = "echo"
+
+    return {
+        "received": True,
+        "source": x_source,
+        "command": command[:200],
+        "action": action,
+        "message": f"[GUARDiA] 명령 수신됨: {command[:100]}",
+        "timestamp": datetime.utcnow().isoformat(),
+    }
+
+
+# ── 외부 SR 인터페이스 (API Key 인증) ────────────────────────────────────────
+
+class ExternalSRCreate(BaseModel):
+    title: str
+    description: str
+    priority: str = "MEDIUM"        # LOW | MEDIUM | HIGH | CRITICAL
+    requester_name: str
+    requester_email: str = ""
+
+
+@router.get("/sr", summary="SR 목록 조회 (API Key read 권한)")
+async def external_list_sr(
+    status_filter: Optional[str] = None,
+    limit: int = 20,
+    db: AsyncSession = Depends(get_db),
+    api_key=Depends(require_scope("read")),
+):
+    q = select(ServiceRequest).limit(limit).order_by(ServiceRequest.created_at.desc())
+    if status_filter:
+        q = q.where(ServiceRequest.status == status_filter)
+    rows = (await db.execute(q)).scalars().all()
+    return [
+        {
+            "id": r.id,
+            "sr_id": r.sr_id,
+            "title": r.title,
+            "status": r.status,
+            "priority": r.priority,
+            "created_at": r.created_at,
+        }
+        for r in rows
+    ]
+
+
+@router.post("/sr", summary="SR 등록 (API Key write 권한)", status_code=201)
+async def external_create_sr(
+    body: ExternalSRCreate,
+    db: AsyncSession = Depends(get_db),
+    api_key=Depends(require_scope("write")),
+):
+    import secrets as _s
+    sr_id = f"EXT-{_s.token_hex(4).upper()}"
+    sr = ServiceRequest(
+        sr_id=sr_id,
+        title=body.title,
+        description=body.description,
+        priority=body.priority,
+        status="RECEIVED",
+        requested_by=f"{body.requester_name} ({body.requester_email})",
+        sr_type="OTHER",
+    )
+    db.add(sr)
+    await db.commit()
+    await db.refresh(sr)
+    return {"id": sr.id, "sr_id": sr.sr_id, "title": sr.title, "status": sr.status, "message": "SR이 등록되었습니다."}