zioinfo-mail/skills/guardia-agent/SKILL.md

name	description
guardia-agent	GUARDiA 프로젝트에서 Python 역방향 에이전트(Reverse Agent) 코드를 작성하거나 수정할 때 사용하는 스킬. 다음 경우에 반드시 먼저 읽어라: - 내부망 중계 PC 에이전트 코드 작성 - asyncio / websockets 기반 역방향 연결 구현 - PostgreSQL psycopg2 DB 조회 코드 - SM 운영 쉘 스크립트 원격 실행 파이프라인 - 에이전트 자동 재연결(Backoff) 로직

GUARDiA 역방향 에이전트 스킬

개념: 역방향 연결 (Reverse Connection)

공공기관 방화벽은 인바운드 차단 → 에이전트가 먼저 아웃바운드로 연결
외부 서버로 전화를 걸어두면 서버가 그 터널을 통해 명령 전달 가능

[내부망 에이전트] ──(Outbound WS)──► [외부 중계 WAS]
                 ◄──(명령 역전달)────

핵심 구현 패턴

무한 재연결 루프 (Backoff)

async def agent_main_loop():
    while True:
        try:
            async with websockets.connect(EXTERNAL_WS_URL) as ws:
                print("연결 성공")
                async for message in ws:
                    await handle_command(ws, json.loads(message))
        except (websockets.ConnectionClosed, OSError):
            print("연결 끊김 — 5초 후 재연결")
            await asyncio.sleep(5)
        except Exception as e:
            print(f"예외: {e}")
            await asyncio.sleep(5)

화이트리스트 기반 명령 분기 (필수)

async def handle_command(ws, packet):
    action = packet.get("action")
    params = packet.get("params", {})
    task_id = packet.get("task_id")
    room_id = packet.get("room_id")

    # 절대로 임의 셸 명령 직접 실행 금지
    # 반드시 정의된 함수만 호출
    if action == "FETCH_MES_QC":
        data = fetch_mes_qc(params.get("date"))
        status = "SUCCESS"
    elif action == "CHECK_INTERNAL_WAS_STATUS":
        data = check_was_health()
        status = "SUCCESS"
    elif action == "CHECK_DISK_SPACE":
        data = check_disk_space()
        status = "SUCCESS"
    else:
        data = {"error": f"허용되지 않은 action: {action}"}
        status = "FAIL"

    await ws.send(json.dumps({
        "event": "TASK_FINISHED",
        "room_id": room_id,
        "task_id": task_id,
        "payload": {"status": status, "data": data}
    }))

DB 조회 표준 (psycopg2)

from psycopg2.extras import RealDictCursor
import datetime

class InfrastructureJsonEncoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj, (datetime.datetime, datetime.date)):
            return obj.isoformat()
        return super().default(obj)

def fetch_mes_qc(work_date: str) -> list:
    """고정 쿼리 + 파라미터 바인딩 (SQL Injection 방지)"""
    conn = psycopg2.connect(**DB_CONFIG)
    try:
        with conn.cursor(cursor_factory=RealDictCursor) as cur:
            sql = """
                SELECT work_date, prc_code, err_msg, reg_date
                FROM tb_mes_wrk_prc_i
                WHERE work_date = %s
                ORDER BY reg_date DESC LIMIT 10
            """
            cur.execute(sql, (work_date,))
            return json.loads(json.dumps(cur.fetchall(), cls=InfrastructureJsonEncoder))
    finally:
        conn.close()

WAS 헬스체크

def check_was_health() -> dict:
    try:
        r = requests.get("http://10.100.10.10:8080/actuator/health", timeout=3)
        return r.json()
    except requests.exceptions.ConnectionError:
        return {"status": "DOWN", "error": "연결 거부"}
    except requests.exceptions.Timeout:
        return {"status": "DOWN", "error": "타임아웃"}

금지 사항

• subprocess.run(user_input) 절대 금지
• 패스워드/API 키 하드코딩 금지 → 환경변수 또는 암호화 설정 파일
• 예외 발생 시 에이전트 프로세스 다운(exit) 금지 → try-except로 감싸고 계속 실행
• 메신저 응답에 내부망 IP 그대로 노출 금지

환경변수 참조 패턴

import os
DB_CONFIG = {
    "host":     os.environ["DB_HOST"],
    "port":     int(os.environ.get("DB_PORT", 5432)),
    "user":     os.environ["DB_USER"],
    "password": os.environ["DB_PASSWORD"],
    "database": os.environ["DB_NAME"],
}
EXTERNAL_WS_URL = os.environ["EXTERNAL_WS_URL"]

Windows 서비스 등록 (운영 배포용)

# NSSM 사용 (Non-Sucking Service Manager)
nssm install GUARDiA-Agent "python" "C:\GUARDiA\src\agent\main.py"
nssm set GUARDiA-Agent AppDirectory "C:\GUARDiA"
nssm start GUARDiA-Agent