Zabbix 인프라 모니터링 구축 – 11. 방화벽 감시: OPNsense 인터페이스·연결 수 모니터링 (SNMP·API)

서버와 네트워크 장비 모니터링을 모두 갖췄다면, 이번 편에서는 방화벽을 다룹니다. OPNsense는 REST API를 제공하기 때문에 SNMP 없이도 다양한 상태 정보를 수집할 수 있습니다.

---

OPNsense REST API 인증

OPNsense API는 API Key/Secret 기반 Basic 인증을 사용합니다.

발급된 Key와 Secret을 Base64로 인코딩해 Authorization 헤더에 포함합니다:

import base64
def api_call(cfg, path, method="GET"):
    credentials = base64.b64encode(
        f"{cfg['key']}:{cfg['secret']}".encode()
    ).decode()
    url = f"https://{cfg['host']}:{cfg['port']}{path}"
    data = b"" if method == "POST" else None
    req = urllib.request.Request(
        url, method=method, data=data,
        headers={"Authorization": f"Basic {credentials}"}
    )
    with urllib.request.urlopen(req, context=ssl_ctx, timeout=15) as resp:
        return json.loads(resp.read().decode())

OPNsense도 자체 서명 인증서를 기본으로 사용하므로 SSL 검증을 비활성화합니다. 포트는 기본 웹 UI와 동일한 443 또는 커스텀 포트(여기서는 8443)를 사용합니다.

---

사이트별 기능 플래그

사이트마다 활성화된 기능이 다를 수 있습니다. 불필요한 API 호출을 피하기 위해 설정에 플래그를 둡니다:

SITES = {
    "hq": {
        "host":        "10.0.91.1",
        "port":        8443,
        "zabbix_host": "opnsense-hq-01",
        "has_gateway": True,   # 게이트웨이 모니터링
        "has_openvpn": True,   # OpenVPN 서버 모니터링
        "has_dhcp":    True,   # DHCP 리스 모니터링
        "services": ["caddy", "suricata", "strongswan",
                     "unbound", "kea-dhcp", "openvpn", "dpinger"],
    },
    "idc": {
        "host":        "10.0.92.1",
        "port":        8443,
        "zabbix_host": "opnsense-idc",
        "has_gateway": False,
        "has_openvpn": False,
        "has_dhcp":    False,
        "services": ["caddy", "suricata", "strongswan", "unbound"],
    },
}

HQ(본사)에는 게이트웨이, OpenVPN, DHCP가 있고, IDC(데이터센터)에는 없습니다. 같은 스크립트로 두 사이트를 처리하면서 플래그로 분기합니다.

---

pf 상태 테이블 수집

pf(Packet Filter)의 상태 테이블은 현재 추적 중인 네트워크 연결 수를 나타냅니다. 테이블이 꽉 차면 새 연결을 거부합니다:

pf = api_call(cfg, "/api/diagnostics/firewall/pfStates")
if pf:
    current = int(pf.get("current", 0))
    limit   = int(pf.get("limit", 1))
    pct     = round(current / limit * 100, 2) if limit else 0
    lines += [
        sl(zh, "opnsense.pf.states.current", current),
        sl(zh, "opnsense.pf.states.limit",   limit),
        sl(zh, "opnsense.pf.states.pct",     pct),
    ]

pct(사용률)가 80% 이상이면 Warning, 90% 이상이면 High 알림을 설정합니다.

---

게이트웨이 상태 수집

인터넷 회선(게이트웨이)의 상태, RTT, 패킷 손실을 수집합니다:

gw = api_call(cfg, "/api/routes/gateway/status")
if gw:
    for item in gw.get("items", []):
        name = item.get("name", "").replace(" ", "_")
        status_str = item.get("status_translated", "down").lower()
        status = 0 if status_str == "online" else 1
        try:
            rtt = float(item.get("delay", "0").replace(" ms", ""))
        except Exception:
            rtt = 0
        try:
            loss = float(item.get("loss", "0").replace(" %", ""))
        except Exception:
            loss = 0
        lines += [
            sl(zh, f"opnsense.gateway.status[{name}]", status),
            sl(zh, f"opnsense.gateway.rtt[{name}]",    rtt),
            sl(zh, f"opnsense.gateway.loss[{name}]",   loss),
        ]

delay와 loss 값은 "12.34 ms", "0.50 %" 형태의 문자열로 반환됩니다. 단위를 제거하고 float으로 변환합니다.

트리거 예시:

# 게이트웨이 다운
last(/opnsense-hq-01/opnsense.gateway.status[WAN_GW]) = 1
# 패킷 손실 10% 초과
last(/opnsense-hq-01/opnsense.gateway.loss[WAN_GW]) > 10
# RTT 100ms 초과
last(/opnsense-hq-01/opnsense.gateway.rtt[WAN_GW]) > 100

---

IPsec 터널 상태 수집

IPsec 사이트 간 VPN 터널의 연결 상태와 트래픽을 수집합니다:

ipsec = api_call(cfg, "/api/ipsec/sessions/searchPhase1", "POST")
if ipsec:
    rows = ipsec.get("rows", [])
    if rows:
        t = rows[0]
        connected = 1 if t.get("connected") else 0
        lines += [
            sl(zh, "opnsense.ipsec.connected",  connected),
            sl(zh, "opnsense.ipsec.bytes_in",   t.get("bytes-in",  0)),
            sl(zh, "opnsense.ipsec.bytes_out",  t.get("bytes-out", 0)),
            sl(zh, "opnsense.ipsec.uptime",     t.get("install-time", 0)),
        ]
    else:
        lines.append(sl(zh, "opnsense.ipsec.connected", 0))

searchPhase1은 IKEv1/IKEv2 Phase 1(메인 터널) 상태를 반환합니다. connected가 false이거나 rows가 비어 있으면 터널이 끊어진 것입니다.

---

서비스 상태 수집

OPNsense에서 실행 중인 서비스의 상태를 한 번에 수집합니다:

svc_result = api_call(cfg, "/api/core/service/search", "POST")
if svc_result:
    svc_map = {}
    for s in svc_result.get("rows", []):
        name    = s.get("name", "")
        running = int(s.get("running", 0))
        # 같은 이름의 서비스가 여러 인스턴스일 수 있으므로 최댓값 사용
        svc_map[name] = max(svc_map.get(name, 0), running)
    for svc in cfg["services"]:
        lines.append(sl(zh, f"opnsense.service[{svc}]", svc_map.get(svc, 0)))

모니터링하는 주요 서비스:

서비스	역할
caddy	리버스 프록시
suricata	IDS/IPS
strongswan	IPsec VPN
unbound	DNS 리졸버
kea-dhcp	DHCP 서버
openvpn	OpenVPN 서버
dpinger	게이트웨이 핑 모니터링

running 값이 0이면 서비스가 중지된 상태입니다. 트리거로 즉시 알림을 보냅니다.

---

Unbound DNS 통계

내부 DNS 리졸버의 쿼리 처리 현황을 수집합니다:

unbound = api_call(cfg, "/api/unbound/diagnostics/stats")
if unbound and unbound.get("status") == "ok":
    data = unbound.get("data", {})
    queries = cachehits = cachemiss = 0
    for thread_data in data.values():
        if not isinstance(thread_data, dict):
            continue
        num = thread_data.get("num", {})
        queries   += int(num.get("queries",   0))
        cachehits += int(num.get("cachehits", 0))
        cachemiss += int(num.get("cachemiss", 0))

응답 구조가 스레드별로 분리되어 있어 전체를 합산합니다. 캐시 히트율이 급락하거나 쿼리 수가 급증하면 DNS 관련 이상 징후를 의심할 수 있습니다.

---

Suricata IDS 알림 수집

침입 탐지 시스템의 알림 건수를 수집합니다:

suricata = api_call(cfg, "/api/ids/service/queryAlerts", "POST")
if suricata:
    total = suricata.get("total", 0)
    drops = sum(1 for r in suricata.get("rows", [])
                if r.get("alert_action") == "drop")
    lines += [
        sl(zh, "opnsense.ids.alerts", total),
        sl(zh, "opnsense.ids.drops",  drops),
    ]

alerts는 탐지 건수, drops는 실제 차단 건수입니다. 급격한 증가는 공격 시도 또는 오탐 규칙 과다를 의미합니다. 절대값보다는 증가 추세에 트리거를 거는 것이 효과적입니다.

---

SSL 인증서 만료일 수집

Let’s Encrypt 인증서의 만료까지 남은 일수를 수집합니다. lastUpdate 타임스탬프 기반으로 계산합니다:

certs = api_call(cfg, "/api/acmeclient/certificates/search", "POST")
if certs:
    for r in certs.get("rows", []):
        if r.get("enabled") != "1":
            continue
        last_update = int(r.get("lastUpdate", 0))
        if last_update:
            age_days    = (now - last_update) / 86400
            expiry_days = round(90 - age_days, 1)  # Let's Encrypt 기본 90일
        else:
            expiry_days = 0
        lines.append(sl(zh, f"opnsense.cert.expiry_days[{r['name']}]", expiry_days))

트리거 예시:

# 인증서 만료 30일 이내 — Warning
last(/opnsense-hq-01/opnsense.cert.expiry_days[도메인명]) < 30
# 만료 7일 이내 — High
last(/opnsense-hq-01/opnsense.cert.expiry_days[도메인명]) < 7

---

LLD: 인증서·OpenVPN 자동 발견

인증서와 OpenVPN 서버 인스턴스는 LLD로 자동 발견합니다:

def do_discovery(cfg):
    zh = cfg["zabbix_host"]
    # 인증서 Discovery
    certs = api_call(cfg, "/api/acmeclient/certificates/search", "POST")
    cert_lld = []
    if certs:
        for r in certs.get("rows", []):
            if r.get("enabled") == "1":
                cert_lld.append({"{#CERTNAME}": r["name"]})
    push_single(zh, "opnsense.cert.discovery", json.dumps(cert_lld))

Discovery를 통해 인증서가 추가되면 자동으로 만료일 아이템이 생성됩니다.

---

crontab 등록

# OPNsense HQ 수집 (2분)
*/2 * * * * /usr/bin/python3 /data/zabbix/scripts/opnsense_collector.py --collect --site hq >> /var/log/zabbix-collector.log 2>&1
# OPNsense IDC 수집 (2분)
*/2 * * * * /usr/bin/python3 /data/zabbix/scripts/opnsense_collector.py --collect --site idc >> /var/log/zabbix-collector.log 2>&1
# OPNsense Discovery (10분)
*/10 * * * * /usr/bin/python3 /data/zabbix/scripts/opnsense_collector.py --discovery --site hq >> /var/log/zabbix-collector.log 2>&1

---

마치며

이번 편에서 다룬 핵심 포인트입니다:

• OPNsense API는 SNMP 없이 Key/Secret Basic 인증으로 접근
• 사이트별 기능 플래그로 불필요한 API 호출 방지
• delay, loss 값이 단위 포함 문자열 — 파싱 필수
• 서비스 상태는 같은 이름 다중 인스턴스 처리 필요 (최댓값 선택)
• Suricata 알림은 절대값보다 증가 추세에 트리거 권장
• 인증서 만료일은 LLD + Item Prototype으로 자동 관리

다음 편에서는 TrueNAS 스토리지 모니터링을 다룹니다.