guardia-itsm/routers/public_sector2.py

203 lines
9.9 KiB
Python

"""
GUARDiA 공공기관 특화 v2 — Gen6
K-CSAP v2·행정망 연동·나라장터 v2·행정전자서명·공공 클라우드·ISP 수립
"""
import uuid
from datetime import datetime, timedelta
from typing import Any, Dict, List, Optional
from fastapi import APIRouter, HTTPException, Query
from pydantic import BaseModel
router = APIRouter(prefix="/api/public2", tags=["Public Sector v2"])
_csap_checks: Dict[str, Dict] = {}
_procurement: Dict[str, Dict] = {}
_admin_net: Dict[str, Dict] = {}
_isp_plans: Dict[str, Dict] = {}
_signatures: Dict[str, Dict] = {}
class CSAPAuditCreate(BaseModel):
institution: str; audit_type: str = "quarterly" # quarterly|annual|special
scope: List[str] = ["all"]
class ProcurementCreate(BaseModel):
title: str; amount: float; category: str
contract_no: str = ""; start_date: str = ""; end_date: str = ""
class AdminNetRequest(BaseModel):
zone: str # admin|internet|dmz
service: str; protocol: str = "https"
approved_by: str
class ISPCreate(BaseModel):
institution: str; fiscal_year: int
total_budget: float; it_budget_ratio: float = 0.05
class ESignRequest(BaseModel):
document_id: str; signer: str; signature_type: str = "gpki" # gpki|accredited|rsa
# ── K-CSAP v2 ────────────────────────────────────────────────────────────
@router.post("/csap/audit")
async def create_csap_audit(audit: CSAPAuditCreate):
aid = f"CSAP-{uuid.uuid4().hex[:8].upper()}"
_csap_checks[aid] = {**audit.model_dump(), "id": aid, "status": "in_progress",
"compliance_rate": 0, "started_at": datetime.utcnow().isoformat()}
return _csap_checks[aid]
@router.get("/csap/audits")
async def list_csap_audits(): return {"audits": list(_csap_checks.values())}
@router.get("/csap/controls")
async def csap_controls():
"""CSAP 보안통제 항목 전체 목록."""
return {"categories": [
{"id": "M", "name": "관리적 보안", "items": 45, "passed": 42, "rate": 93.3},
{"id": "P", "name": "물리적 보안", "items": 20, "passed": 19, "rate": 95.0},
{"id": "T", "name": "기술적 보안", "items": 80, "passed": 73, "rate": 91.3},
], "total": 145, "passed": 134, "overall_rate": 92.4}
@router.get("/csap/report/{aid}")
async def csap_report(aid: str):
audit = _csap_checks.get(aid)
if not audit: raise HTTPException(404)
return {**audit, "compliance_rate": 92.4,
"findings": [{"control": "T-3.2", "status": "미흡", "recommendation": "패스워드 정책 강화"},
{"control": "M-1.5", "status": "보완", "recommendation": "보안 교육 주기 단축"}],
"next_audit": (datetime.utcnow() + timedelta(days=90)).isoformat()}
@router.get("/csap/gap-analysis")
async def csap_gap_analysis(institution: str = Query(...)):
return {"institution": institution, "gap_items": [
{"control": "T-5.1", "current_state": "미구현", "target": "구현", "priority": "high"},
{"control": "M-2.3", "current_state": "부분구현", "target": "완전구현", "priority": "medium"},
], "improvement_plan": "3개월 내 2개 항목 개선 계획"}
@router.post("/csap/self-check")
async def csap_self_check(institution: str, category: str = "all"):
return {"institution": institution, "category": category,
"checked_at": datetime.utcnow().isoformat(),
"score": 92.4, "grade": "우수",
"action_items": 3, "status": "completed"}
# ── 나라장터 v2 ────────────────────────────────────────────────────────────
@router.post("/g2b/procurement")
async def register_procurement(proc: ProcurementCreate):
pid = f"G2B-{uuid.uuid4().hex[:8].upper()}"
_procurement[pid] = {**proc.model_dump(), "id": pid, "status": "registered",
"registered_at": datetime.utcnow().isoformat()}
return _procurement[pid]
@router.get("/g2b/procurement")
async def list_procurement(status: Optional[str] = None):
procs = list(_procurement.values())
if status: procs = [p for p in procs if p.get("status") == status]
return {"procurements": procs, "total": len(procs)}
@router.get("/g2b/search")
async def search_g2b(keyword: str, category: str = "IT", page: int = 1):
return {"keyword": keyword, "category": category, "page": page,
"results": [
{"id": "G2B-001", "title": f"[{category}] {keyword} 시스템 구축", "amount": 150000000,
"deadline": "2026-07-15", "status": "공고중"},
{"id": "G2B-002", "title": f"{keyword} 유지보수 용역", "amount": 48000000,
"deadline": "2026-07-20", "status": "공고중"},
], "total": 2}
@router.get("/g2b/contract/{cid}")
async def get_contract(cid: str):
return {"contract_id": cid, "title": "GUARDiA ITSM 유지보수", "amount": 48000000,
"period": "2026-01-01 ~ 2026-12-31", "status": "계약중", "vendor": "지오정보기술"}
@router.post("/g2b/delivery-check")
async def delivery_check(contract_id: str, items: List[Dict[str, Any]]):
return {"contract_id": contract_id, "items_checked": len(items),
"status": "검수완료", "checked_at": datetime.utcnow().isoformat(),
"inspector": "담당자", "next_step": "세금계산서 발행"}
# ── 행정망 연동 관리 ─────────────────────────────────────────────────────
@router.post("/admin-net/request")
async def request_admin_net(req: AdminNetRequest):
rid = f"NET-{uuid.uuid4().hex[:8].upper()}"
_admin_net[rid] = {**req.model_dump(), "id": rid, "status": "pending",
"requested_at": datetime.utcnow().isoformat()}
return _admin_net[rid]
@router.get("/admin-net/topology")
async def admin_net_topology():
return {"zones": [
{"name": "행정망", "type": "admin", "services": ["ITSM", "CMDB"], "firewall_rules": 24},
{"name": "인터넷망", "type": "internet", "services": ["Homepage"], "firewall_rules": 12},
{"name": "DMZ", "type": "dmz", "services": ["Manager API"], "firewall_rules": 8},
], "connections": [
{"from": "admin", "to": "dmz", "protocol": "https", "status": "active"},
{"from": "internet", "to": "dmz", "protocol": "https", "status": "active"},
]}
@router.get("/admin-net/firewall-rules")
async def firewall_rules(zone: Optional[str] = None):
rules = [
{"id": "FW-001", "zone": "admin", "src": "10.0.0.0/8", "dst": "any", "port": 443, "action": "allow"},
{"id": "FW-002", "zone": "internet", "src": "any", "dst": "DMZ", "port": 443, "action": "allow"},
]
if zone: rules = [r for r in rules if r["zone"] == zone]
return {"rules": rules, "total": len(rules)}
# ── 행정전자서명 (GPKI) ────────────────────────────────────────────────────
@router.post("/esign/request")
async def esign_request(req: ESignRequest):
sid = f"SIG-{uuid.uuid4().hex[:8].upper()}"
_signatures[sid] = {**req.model_dump(), "id": sid, "status": "pending",
"requested_at": datetime.utcnow().isoformat()}
return _signatures[sid]
@router.post("/esign/verify")
async def esign_verify(signature_id: str):
sig = _signatures.get(signature_id)
if not sig: raise HTTPException(404)
return {"signature_id": signature_id, "valid": True, "signer": sig.get("signer"),
"signed_at": datetime.utcnow().isoformat(), "certificate": "행정기관인증서"}
# ── ISP 수립 지원 v2 ──────────────────────────────────────────────────────
@router.post("/isp")
async def create_isp(isp: ISPCreate):
iid = f"ISP-{uuid.uuid4().hex[:8].upper()}"
_isp_plans[iid] = {**isp.model_dump(), "id": iid, "status": "draft",
"it_budget": isp.total_budget * isp.it_budget_ratio,
"created_at": datetime.utcnow().isoformat()}
return _isp_plans[iid]
@router.get("/isp")
async def list_isp(): return {"plans": list(_isp_plans.values())}
@router.get("/isp/{iid}/roadmap")
async def isp_roadmap(iid: str):
isp = _isp_plans.get(iid)
if not isp: raise HTTPException(404)
return {"isp_id": iid, "roadmap": [
{"quarter": "Q1", "projects": ["ITSM 고도화"], "budget": 50000000},
{"quarter": "Q2", "projects": ["보안 강화"], "budget": 30000000},
{"quarter": "Q3", "projects": ["DR 구축"], "budget": 40000000},
{"quarter": "Q4", "projects": ["사용자 교육"], "budget": 10000000},
]}
# ── 공공 클라우드 (K-Cloud) ────────────────────────────────────────────────
@router.get("/kcloud/status")
async def kcloud_status():
return {"provider": "NCloud (공공)", "region": "kr-pub-1",
"services_deployed": 3, "cost_this_month": 1240000,
"compliance": "CSAP 인증 완료", "availability": "99.98%"}
@router.get("/kcloud/pricing")
async def kcloud_pricing(resource_type: str = "compute"):
pricing = {
"compute": [{"spec": "2vCPU/4GB", "price_hour": 85, "price_month": 61200}],
"storage": [{"spec": "100GB SSD", "price_month": 15000}],
"network": [{"spec": "공인IP", "price_month": 6600}],
}
return {"resource_type": resource_type, "pricing": pricing.get(resource_type, [])}
@router.get("/public2/health")
async def health():
return {"status": "healthy", "csap_audits": len(_csap_checks),
"procurement": len(_procurement), "signatures": len(_signatures)}