Mini-SOAR 엔진의 계층 구조와 파이썬 모듈의 처리 흐름

지난 <SOAR 구축 실습: Snort 경보 기반 자동 침해대응 플레이북 엔진>에서는 공격 탐지부터 차단·알림·티켓 발행까지 이어지는 전체 파이프라인을 구축하고 End-to-End로 검증하는 과정을 다뤘다.

그 때는 인프라 구성과 통합 흐름에 초점을 맞췄다면, 이번에는 그 파이프라인의 중심부에 해당하는 SOAR 엔진 자체의 내부 설계와 작동 원리를 자세히 살펴본다.

1. Mini-SOAR 플레이북 엔진의 계층 구조

 

Mini-SOAR 엔진은 네 개의 계층(모듈)으로 분리했다.

진입점(app.py)에서 경보를 수신하고, 파싱 계층(snort_parser.py)이 원본 문자열을 구조화된 데이터로 바꾸고, 매칭 계층(playbook_engine.py)이 어떤 플레이북을 실행할지 판단하고, 실행 계층(actions/)이 실제 대응 동작을 수행하는 구조이다.

각 모듈은 바로 아래 계층의 모듈만 호출하며, 각자 맡은 일 외에는 몰라도 된다.

 

1️⃣ 각 계층의 책임 경계

🔵 진입점 (app.py) — Flask 애플리케이션 그 자체로, 외부에서 들어오는 HTTP 요청을 받아 원본 경보 문자열을 꺼내고, 아래 세 모듈을 순서대로 호출한 뒤 결과를 JSON 응답으로 되돌려준다. 비즈니스 로직을 직접 수행하지 않고 오케스트레이션만 담당한다. 

 

🔵 파싱 계층 (snort_parser.py) — Snort가 보내는 비정형 경보 문자열을 구조화된 데이터클래스로 변환한다. 파싱 결과는 @dataclass로 표현한다. raw 필드에 원본 문자열을 보존해 Notion 티켓에 원문을 첨부할 수 있게 했다.

@dataclass
class SnortAlert:
    raw: str
    signature: str = ""
    sig_id: str = ""              # gid:sid:rev
    attacker_ip: str = ""
    attacker_port: Optional[int] = None
    target_ip: str = ""
    target_port: Optional[int] = None
    protocol: str = ""
    priority: int = 3
    severity: str = "low"
    classification: str = ""
    timestamp: str = ""

 

Snort의 출력 포맷은 설정(alert_fast vs alert_full)에 따라 다르다.

Fast 포맷에는 {TCP} 같은 프로토콜 마커가 있어 우선 매칭하고, 없으면 IP 패턴으로 폴백한다.

# Fast format (단일 라인): {TCP} 1.2.3.4:port -> 5.6.7.8:port
proto_match = re.search(
    r'\{(\w+)\}\s*([\d.]+)(?::(\d+))?\s*->\s*([\d.]+)(?::(\d+))?', raw
)

# Full format (다중 라인, {PROTO} 없음)
ip_match = re.search(
    r'([\d.]+)(?::(\d+))?\s*->\s*([\d.]+)(?::(\d+))?', raw
)

Snort의 Priority는 숫자가 낮을수록 심각도가 높다(1=Critical).

YAML 플레이북에서 severity_in: [high, medium]처럼 사람이 읽기 쉬운 문자열로 조건을 쓸 수 있도록, 매핑 테이블을 모듈 상단 상수로 분리해 변환한다.

_PRIORITY_SEVERITY = {1: "high", 2: "medium", 3: "low"}
alert.severity = _PRIORITY_SEVERITY.get(alert.priority, "low")

 

🔵 매칭 계층 (playbook_engine.py) — 파싱된 경보를 YAML로 선언된 플레이북 조건과 비교한다. 대응 로직이 코드가 아니라 데이터에 담겨 있다는 점이 이 계층의 핵심이다. 플레이북 로드는 별도 메서드로 분리했는데, yaml.load()는 임의의 Python 객체를 역직렬화할 수 있어 YAML 조작을 통한 코드 실행 취약점이 생기므로 기본 자료형만 허용하는 safe_load를 쓴다. 로드를 메서드로 떼어 두면 POST /api/playbooks/reload 엔드포인트로 서버 재시작 없이 플레이북을 갱신할 수 있다.

def load_playbooks(self) -> None:
    self.playbooks = []
    for fname in sorted(os.listdir(self.playbooks_dir)):
        if not fname.endswith((".yml", ".yaml")):
            continue
        pb = yaml.safe_load(f)
        if pb.get("enabled", True):
            self.playbooks.append(pb)

 

조건 평가는 AND 로직이다. "Shellshock 패턴이고 AND 심각도가 high/medium"처럼 복합 조건으로 오탐을 줄이기 위해, 조건이 하나라도 어긋나면 즉시 탈락시킨다.

def _check_conditions(self, conditions: Dict, alert: SnortAlert) -> bool:
    for key, value in conditions.items():
        if not self._eval_condition(key, value, alert):
            return False   # 하나라도 False면 즉시 탈락
    return True

 

조건 키는 if 분기로 확장 가능하게 두었다. 알 수 없는 조건 키를 만나면 False가 아니라 True(무시)로 처리하는데, 이는 새 조건 키를 YAML에 먼저 쓰고 엔진 배포가 늦어지는 경우 플레이북 전체가 멈추는 것을 막기 위한 선택이다. 운영 연속성을 우선한 설계이며, 보안을 더 중시하는 환경에서는 False 처리가 안전할 수 있다.

def _eval_condition(self, key, value, alert) -> bool:
    if key == "signature_contains": ...
    if key == "signature_regex": ...
    if key == "severity": ...
    if key == "severity_in": ...
    if key == "priority_lte": ...
    if key == "attacker_ip": ...
    if key == "protocol": ...
    if key == "classification_contains": ...
    logger.warning("Unknown condition key: %s", key)
    return True   # 알 수 없는 조건은 무시 (차단하지 않음)

 

템플릿 변수 치환에는 format_map()을 쓴다. str.format(**vars)와 달리 누락된 키에 대해 KeyError 대신 원본 문자열을 유지하기 쉬워, {attacker_ip} 같은 자리표시자를 실제 경보 데이터로 안전하게 바꿔 Slack 메시지와 Notion 티켓 제목을 동적으로 생성한다.

def _render_dict(d: Dict, vars: Dict) -> Dict:
    for k, v in d.items():
        if isinstance(v, str):
            out[k] = v.format_map(vars)

🔵 실행 계층 (actions/) — 매칭된 플레이북이 지정한 액션을 실제로 수행한다. 액션 실행은 중앙 디스패치 패턴으로 묶는데, 매칭 엔진이 block.py를 직접 import하면 액션이 늘 때마다 엔진을 고쳐야 하지만 dispatch_action()을 단일 진입점으로 두면 새 액션 추가 시 actions/__init__.py만 수정하면 되고 엔진은 그대로 둘 수 있다(개방-폐쇄 원칙).

 

 

def dispatch_action(action_type, action_def, alert) -> Tuple[bool, Dict]:
    if action_type == "block_ip": ...
    if action_type == "slack_notify": ...
    if action_type == "notion_ticket": ...
    if action_type == "log": ...

 

가장 중요한 block_ip()는 이중 방어 구조를 갖는다.

subprocess.run(list) 방식으로 셸 해석을 우회하고, 정규식으로 IP 포맷을 선검증한 뒤 SSH로 IP 문자열만 전달한다.

IPS 측 soar_iptables.sh도 자체 IP 검증을 수행하므로, SOAR와 IPS 양쪽에서 검증이 이뤄진다.

_IP_RE = re.compile(r'^\d{1,3}(?:\.\d{1,3}){3}$')

def block_ip(ip: str) -> bool:
    if not _IP_RE.match(ip):
        logger.error("block_ip: invalid IP format: %s", ip)
        return False
    cmd = ["ssh", "-i", ssh_key, ..., f"root@{fwips_host}", ip]
    subprocess.run(cmd, timeout=15)

2. 경보 전송부터 자동 차단까지의 전체 흐름

계층 구조가 정적인 뼈대라면, 실제 경보 하나가 들어왔을 때 그 뼈대를 따라 데이터가 어떻게 이동하는지는 동적인 흐름이다.

 

1️⃣ Flask 데코레이터가 요청을 함수로 연결하는 방식

Snort가 있는 IPS 서버가 http://192.168.100.35:5000/api/alerts로 HTTP POST를 보낸다.

이 요청이 SOAR 서버에 도착했을 때, 어떤 함수가 실행될지를 결정하는 것이 @app.route('/api/alerts', methods=['POST']) 데코레이터다.

Flask는 내부에 URL과 함수의 매핑 테이블을 갖고 있고, 데코레이터가 붙는 순간 receive_alert 함수가 /api/alerts 경로에 등록된다.

이후 요청이 들어오면 Flask가 경로를 조회해 해당 함수를 호출한다.

methods=['POST']를 명시했기 때문에 GET 등 다른 메서드로 같은 경로를 호출하면 405 응답이 돌아간다.

경보 수신이라는 목적상 POST만 허용하는 것이 자연스럽다.

 

진입점 함수는 요청을 받으면 먼저 인시던트 ID를 채번한다.

INC-20260619-0001 형식으로, 날짜가 들어가 로그 검색 시 범위 필터링이 쉽고 4자리 순번으로 하루 9,999건까지 처리할 수 있다.

UUID 대신 사람이 읽을 수 있는 ID라 티켓 제목과 Slack 메시지에서 식별이 쉽다.

def _next_id() -> str:
    global _counter
    _counter += 1
    date = datetime.now(timezone.utc).strftime("%Y%m%d")
    return f"INC-{date}-{_counter:04d}"

 

IPS 서버가 경보를 보내는 방식은 환경마다 다르다.

단순 curl로 텍스트를 직접 보내거나, rsyslog가 JSON으로 보내기도 한다.

어느 쪽이든 수용하도록 Content-Type을 분기해, IPS 측 설정 변경 없이 연동할 수 있게 했다.

ct = request.content_type or ""
if "json" in ct:
    body = request.get_json(force=True, silent=True) or {}
    raw = body.get("raw") or body.get("alert") or body.get("message") or ""
else:
    raw = request.get_data(as_text=True)

2️⃣ receive_alert 함수 내부의 호출

라우팅이 receive_alert를 호출하면, 그 함수 안에서 위의 세 모듈이 순서대로 호출된다.

다이어그램에서 보라색 점선 박스로 묶인 부분이 이 함수의 몸통이며, 내부에서 ①②③으로 표시된 세 번의 호출이 일어난다.

🔵 ① parse_snort_alert(raw) — 요청 본문에서 꺼낸 원본 경보 문자열 raw를 파서에 넘긴다. 파서는 정규식으로 공격자 IP, 시그니처, 심각도를 추출해 SnortAlert 객체로 되돌려준다. 이 시점부터 경보는 문자열이 아니라 속성으로 접근 가능한 구조화된 데이터가 된다.

🔵 ② engine.match(alert) — 파싱된 alert 객체를 매칭 엔진에 넘긴다. 엔진은 로드된 모든 YAML 플레이북을 순회하며 각 플레이북의 조건 블록을 경보와 비교하고, 부합하는 플레이북 목록을 반환한다. 조건이 하나도 맞지 않으면 이후 단계 없이 여기서 흐름이 끝난다.

🔵 ③ engine.execute → dispatch_action() — 매칭된 플레이북이 있으면, 그 플레이북이 지정한 액션들을 실행한다. 여기서 dispatch_action()이 액션 타입 문자열(block_ip, slack_notify, notion_ticket)을 보고 알맞은 핸들러 함수로 분기시킨다. 이 분기 지점이 있어서 매칭 엔진은 무엇을 실행할지만 알면 되고, 어떻게 실행할지는 실행 계층에서 책임진다.

 

3️⃣ 액션 분기와 실제 차단

dispatch_action()에서 갈라진 세 액션은 각각 독립적으로 수행된다.

block_ip()는 SSH로 방화벽에 차단을 지시하고, slack_notify()는 Webhook으로 알림을 쏘고, notion_ticket()은 API로 티켓을 만든다.

block_ip()는 subprocess.run(ssh ...)으로 IPS 서버에 SSH 접속을 시도한다.

여기서 SOAR가 보내는 것은 명령어 전체가 아니라 차단할 IP 문자열이고, 실제로 어떤 명령이 실행될지는 SOAR가 아니라 IPS 서버가 결정한다.

4️⃣ IPS 서버가 명령을 강제하는 구조

SSH 접속이 IPS 서버에 도달하면, authorized_keys에 걸린 command= 제한이 작동한다.

이 옵션은 SOAR의 공개키로 접속한 세션이 무슨 명령을 보내든 무조건 soar_iptables.sh 스크립트만 실행하도록 강제한다.

SOAR가 임의의 셸 명령을 보내려 해도, IPS 서버는 그것을 무시하고 지정된 차단 스크립트만 돌린다.

# IPS 서버의 /root/.ssh/authorized_keys
command="/usr/local/bin/soar_iptables.sh",no-port-forwarding,no-X11-forwarding ssh-rsa AAAA...

 

soar_iptables.sh는 전달받은 IP가 올바른 IPv4 형식인지 자체적으로 한 번 더 검증한 뒤, 유효하면 iptables -I FORWARD -s <IP> -j DROP을 실행한다.

SSH 접속 시 클라이언트가 보낸 값은 $SSH_ORIGINAL_COMMAND 환경변수에 담기므로, 여기서 IP 패턴만 추출해 검증한다.

#!/bin/bash
IP=$(echo "$SSH_ORIGINAL_COMMAND" | grep -oP '^\d+\.\d+\.\d+\.\d+$')
if [ -n "$IP" ]; then
    /sbin/iptables -I FORWARD -s "$IP" -j DROP
    echo "Blocked $IP in FORWARD chain"
else
    echo "Invalid command" >&2
    exit 1
fi

 

이 구조는 신뢰 경계를 SOAR가 아니라 IPS 서버에 두고 있다.

SOAR 서버가 침해당해 공격자에게 넘어가더라도 할 수 있는 건 IP 하나를 차단하는 것뿐이다.

자동화 주체에 방화벽 전체 권한을 넘기지 않고, 필요한 동작 하나만 위임하는 최소 권한 원칙이 SSH 계층에서 구조적으로 구현되어 있다.

 

3. 정리

계층을 나눈 정적 설계 덕분에 각 모듈은 독립적으로 테스트하고 확장할 수 있고, 한 방향으로만 흐르는 동적 설계 덕분에 경보 처리 경로를 예측 가능하게 추적할 수 있다.

새 공격 유형에 대응하려면 YAML 플레이북을 추가하면 되고, 새 대응 수단이 필요하면 actions/에 모듈 하나를 붙이고 dispatch_action()에 분기 한 줄만 더하면 된다.