ESET Threat Intelligence 로그 수집

다음에서 지원:

이 문서에서는 Google Cloud Storage V2, Cloud Run 함수, Cloud Scheduler를 사용하여 ESET 위협 인텔리전스 로그를 Google Security Operations에 수집하는 방법을 설명합니다.

ESET 위협 인텔리전스 (ETI)는 기존 위협 또는 새롭게 등장하는 위협에 대한 증거 기반 정보와 실행 가능한 조언을 제공합니다. ETI 서비스는 조직 또는 고객을 위협할 수 있는 악성 소프트웨어 또는 활동에 대해 경고합니다. 이 서비스는 APT IoC, 봇넷 C&C 및 타겟, 악성 도메인, IP, URL, 파일, 피싱 URL, 랜섬웨어, Android 위협을 비롯한 위협 인텔리전스 데이터를 STIX 2.1 형식의 TAXII 2.1 피드를 통해 제공합니다.

시작하기 전에

다음 기본 요건이 충족되었는지 확인합니다.

  • Google SecOps 인스턴스
  • 다음 API가 사용 설정된 Google Cloud 프로젝트
    • Cloud Storage API
    • Cloud Run Functions API
    • Cloud Scheduler API
    • Cloud Pub/Sub API
  • Google Cloud Storage 버킷, Cloud Run 함수, Pub/Sub 주제, Cloud Scheduler 작업을 만들고 관리할 수 있는 권한
  • Google Cloud Storage 버킷의 IAM 정책을 관리할 수 있는 권한
  • 활성 상태의 ESET Threat Intelligence 구독
  • https://eti.eset.com의 ESET 위협 인텔리전스 포털에 대한 액세스

Google Cloud Storage 버킷 만들기

  1. Google Cloud 콘솔로 이동합니다.
  2. 프로젝트를 선택하거나 새 프로젝트를 만듭니다.
  3. 탐색 메뉴에서 Cloud Storage> 버킷으로 이동합니다.
  4. 버킷 만들기를 클릭합니다.

  5. 다음 구성 세부정보를 제공합니다.

    설정 값
    버킷 이름 지정 전역적으로 고유한 이름 (예: eset-ti-logs)을 입력합니다.
    위치 유형 필요에 따라 선택 (리전, 이중 리전, 멀티 리전)
    위치 위치를 선택합니다 (예: us-central1).
    스토리지 클래스 Standard (자주 액세스하는 로그에 권장)
    액세스 제어 균일 (권장)
    보호 도구 선택사항: 객체 버전 관리 또는 보관 정책 사용 설정

  6. 만들기를 클릭합니다.

ESET 위협 인텔리전스 TAXII 사용자 인증 정보 수집

Cloud Run 함수가 위협 인텔리전스 데이터를 가져오도록 하려면 ETI 포털에서 TAXII 피드를 활성화하고 TAXII 사용자 인증 정보를 생성해야 합니다.

TAXII 피드 활성화

  1. https://eti.eset.com에서 ESET 위협 인텔리전스 포털에 로그인합니다.
  2. 기본 메뉴에서 데이터 피드로 이동합니다.
  3. 활성화할 데이터 피드 옆에 있는 점 3개 아이콘을 클릭합니다.
  4. 피드 사용 설정을 선택합니다.

  5. Google SecOps에 수집하려는 각 피드에 대해 3~4단계를 반복합니다.

TAXII 사용자 인증 정보 생성

  1. ESET 위협 인텔리전스 포털에서 관리 설정 > 액세스 사용자 인증 정보로 이동합니다.
  2. TAXII 사용자 인증 정보 생성을 클릭합니다.

  3. 대화상자가 나타나면 다음 값을 복사하여 저장합니다.

    • 사용자 이름: TAXII 사용자 이름
    • 비밀번호: TAXII 비밀번호

TAXII 피드 세부정보 기록

피드를 활성화하고 사용자 인증 정보를 생성한 후 수집하려는 각 피드에 대해 다음 정보를 기록합니다.

  1. ESET Threat Intelligence 포털에서 데이터 피드로 이동합니다.
  2. 활성화된 피드 옆에 있는 점 3개 아이콘을 클릭합니다.
  3. 데이터 피드 세부정보 표시를 선택합니다.

  4. 측면 패널에서 다음 값을 확인합니다.

    • TAXII 피드 이름: 피드 식별자 (예: botnet stix 2.1)
    • TAXII 2 ID: 컬렉션 ID (예: 0abb06690b0b47e49cd7794396b76b20)
    • TAXII 2 피드 URL: 전체 컬렉션 URL

사용 가능한 TAXII 피드

  • ESET Threat Intelligence는 다음과 같은 TAXII 2.1 피드를 제공합니다.

    피드 이름 TAXII 피드 이름 컬렉션 ID
    Android 정보 유출 피드 androidinfostealer stix 2.1 9ee501cde0c44d6db4ae995fead1a7c8
    Android 위협 피드 androidthreats stix 2.1 daf3de8fab144552a1cb5af054ed07ee
    APT IoC apt stix 2.1 97e3eb74ae5f46dd9e22f677a6938ee7
    봇넷 피드 botnet stix 2.1 0abb06690b0b47e49cd7794396b76b20
    봇넷 - C&C botnet.cc stix 2.1 d1923a526e8f400dbb301259240ee3d5
    봇넷 - 타겟 botnet.target stix 2.1 61b6e4f9153e411ca7a9982a2c6ae788
    암호화폐 사기 피드 cryptoscam stix 2.1 2c183ce9551a43338c6cc2ed7c2a704d
    도메인 피드 domain stix 2.1 a34aa0a4f9de419582a883863503f9c4
    eCrime IoC 피드 ecrime stix 2.1 08059376eac84ec4a076cfd682493f91
    IP 피드 ip stix 2.1 baaed2a92335418aa753fe944e13c23a
    악성 이메일 첨부파일 emailattachments stix 2.1 c0d56cf7f81d482eb97fd46beaa4bae0
    악성 파일 피드 file stix 2.1 ee6a153ed77e4ec3ab21e76cc2074b9f
    피싱 URL 피드 phishingurl stix 2.1 d0a6c0f962dd4dd2b3eeb96b18612584
    PUA 애드웨어 파일 피드 puaadware stix 2.1 d1bfc81202fc4c6599326771ec2da41d
    PUA 이중 용도 앱 파일 피드 puadualapps stix 2.1 970a7d0039ac4668addf058cd9feb953
    랜섬웨어 피드 랜섬웨어 stix 2.1 8d3490d688ce4a989aee9af5c680d8bf
    사기 URL 피드 scamurl stix 2.1 2130adc3c67c43f9a3664b187931375e
    스미싱 피드 smishing stix 2.1 330ad7d0c736476babe5e49077b96c95
    SMS 사기 피드 smsscam stix 2.1 6e20217a2e1246b8ab11be29f759f716
    URL 피드 url stix 2.1 1d3208c143be49da8130f5a66fd3a0fa

Cloud Run 함수의 서비스 계정 만들기

  1. Google Cloud 콘솔에서 IAM 및 관리자 > 서비스 계정으로 이동합니다.
  2. 서비스 계정 만들기를 클릭합니다.

  3. 다음 구성 세부정보를 제공합니다.

    • 서비스 계정 이름: eset-ti-collector을 입력합니다.
    • 서비스 계정 설명: Service account for ESET Threat Intelligence Cloud Run function to write STIX objects to GCS를 입력합니다.

  4. 만들고 계속하기를 클릭합니다.

  5. 이 서비스 계정에 프로젝트에 대한 액세스 권한 부여 섹션에서 다음 역할을 추가합니다.

    1. 역할 선택을 클릭하고 스토리지 객체 관리자를 검색하여 선택합니다.
    2. 다른 역할 추가를 클릭하고 Cloud Run 호출자를 검색하여 선택합니다.

  6. 계속을 클릭합니다.

  7. 완료를 클릭합니다.

Google Cloud Storage 버킷에 대한 IAM 권한 부여

  1. Cloud Storage> 버킷으로 이동합니다.
  2. 버킷 이름 (예: eset-ti-logs)을 클릭합니다.
  3. 권한 탭으로 이동합니다.
  4. 액세스 권한 부여를 클릭합니다.

  5. 다음 구성 세부정보를 제공합니다.

    • 주 구성원 추가: 서비스 계정 이메일 (예: eset-ti-collector@PROJECT_ID.iam.gserviceaccount.com)을 입력합니다.
    • 역할 할당: 스토리지 객체 관리자를 선택합니다.

  6. 저장을 클릭합니다.

게시/구독 주제 만들기

Cloud Scheduler에서 메시지를 게시하면 Pub/Sub 주제가 Cloud Run 함수를 트리거합니다.

  1. Google Cloud 콘솔에서 Pub/Sub > 주제로 이동합니다.
  2. 주제 만들기를 클릭합니다.
  3. 다음 구성 세부정보를 제공합니다.
    • 주제 ID: eset-ti-trigger를 입력합니다.
    • 기본 구독 추가: 선택된 상태로 유지
  4. 만들기를 클릭합니다.

Cloud Run 함수 만들기

  1. Google Cloud 콘솔에서 Cloud Run 함수로 이동합니다.
  2. 함수 만들기를 클릭합니다.

  3. 다음 구성 세부정보를 제공합니다.

    설정 값
    환경 2세대
    함수 이름 eset-ti-collector
    리전 GCS 버킷과 동일한 리전을 선택합니다.
    트리거 유형 Cloud Pub/Sub
    Pub/Sub 주제 eset-ti-trigger
    할당된 메모리 512MB
    제한 시간 540초
    런타임 서비스 계정 eset-ti-collector

  4. 다음을 클릭합니다.

  5. 런타임을 Python 3.12로 설정합니다.

  6. 진입점을 main로 설정합니다.

  7. requirements.txt 파일에 다음 종속 항목을 추가합니다.

    functions-framework==3.*
google-cloud-storage==2.*
urllib3==2.*

  8. main.py 파일에 다음 코드를 붙여넣습니다.

    import functions_framework
import json
import os
import logging
import time
import urllib3
from datetime import datetime, timedelta, timezone
from google.cloud import storage

logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

HTTP = urllib3.PoolManager(retries=False)
storage_client = storage.Client()

API_ROOT = "https://taxii.eset.com/taxii2/643f4eb5-f8b7-46a3-a606-6d61d5ce223a"
TAXII_CONTENT_TYPE = "application/taxii+json;version=2.1"

def _load_state(bucket_name: str, state_key: str, lookback_hours: int) -> str:
    """Return ISO8601 checkpoint (UTC)."""
    try:
        bucket = storage_client.bucket(bucket_name)
        blob = bucket.blob(state_key)
        if blob.exists():
            state_data = blob.download_as_text()
            state = json.loads(state_data)
            ts = state.get("last_poll_time")
            if ts:
                logger.info(f"Loaded state: {ts}")
                return ts
    except Exception as e:
        logger.warning(f"State read error: {e}")
    default_ts = (
        datetime.now(timezone.utc) - timedelta(hours=lookback_hours)
    ).strftime("%Y-%m-%dT%H:%M:%S.000Z")
    logger.info(f"No previous state found, using lookback: {default_ts}")
    return default_ts

def _save_state(bucket_name: str, state_key: str, ts: str) -> None:
    """Persist the checkpoint to GCS."""
    bucket = storage_client.bucket(bucket_name)
    blob = bucket.blob(state_key)
    blob.upload_from_string(
        json.dumps({"last_poll_time": ts}),
        content_type="application/json",
    )
    logger.info(f"Saved state: {ts}")

def _fetch_objects(
    username: str,
    password: str,
    collection_id: str,
    added_after: str,
    max_records: int,
) -> list:
    """Query TAXII 2.1 collection objects with pagination."""
    url = f"{API_ROOT}/collections/{collection_id}/objects/"
    headers = urllib3.make_headers(basic_auth=f"{username}:{password}")
    headers["Accept"] = TAXII_CONTENT_TYPE
    headers["User-Agent"] = "Chronicle-ESET-TI-GCS/1.0"

    all_objects = []
    params = {"added_after": added_after}

    while True:
        qs = "&".join(f"{k}={v}" for k, v in params.items())
        request_url = f"{url}?{qs}" if qs else url

        for attempt in range(3):
            try:
                resp = HTTP.request("GET", request_url, headers=headers)
                break
            except Exception as e:
                wait = 2 ** (attempt + 1)
                logger.warning(f"Request error: {e}, retrying in {wait}s")
                time.sleep(wait)
        else:
            raise RuntimeError("Exceeded retry budget for TAXII API")

        if resp.status == 401:
            raise RuntimeError("Authentication failed: check TAXII credentials")
        if resp.status == 404:
            raise RuntimeError(
                f"Collection not found: {collection_id}"
            )
        if resp.status not in (200, 206):
            raise RuntimeError(
                f"TAXII API error {resp.status}: {resp.data[:500]}"
            )

        body = json.loads(resp.data.decode("utf-8"))
        objects = body.get("objects", [])
        all_objects.extend(objects)
        logger.info(
            f"Fetched {len(objects)} objects (total: {len(all_objects)})"
        )

        if len(all_objects) >= max_records:
            logger.info(f"Reached max_records limit: {max_records}")
            all_objects = all_objects[:max_records]
            break

        more = body.get("more", False)
        next_param = body.get("next")
        if more and next_param:
            params = {"added_after": added_after, "next": next_param}
        else:
            break

    return all_objects

@functions_framework.cloud_event
def main(cloud_event):
    """Cloud Run function entry point triggered by Pub/Sub."""
    bucket_name = os.environ["GCS_BUCKET"]
    prefix = os.environ.get("GCS_PREFIX", "eset-ti")
    state_key = os.environ.get("STATE_KEY", "eset-ti/state.json")
    username = os.environ["TAXII_USERNAME"]
    password = os.environ["TAXII_PASSWORD"]
    collection_id = os.environ["COLLECTION_ID"]
    max_records = int(os.environ.get("MAX_RECORDS", "10000"))
    lookback_hours = int(os.environ.get("LOOKBACK_HOURS", "48"))

    try:
        last_poll = _load_state(bucket_name, state_key, lookback_hours)
        objects = _fetch_objects(
            username, password, collection_id, last_poll, max_records
        )

        if not objects:
            logger.info("No new STIX objects found")
            return "No new objects", 200

        now_str = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")
        blob_path = (
            f"{prefix}/eset_ti_{collection_id}_{now_str}.json"
        )
        ndjson_body = "\n".join(
            json.dumps(obj, separators=(",", ":")) for obj in objects
        )

        bucket = storage_client.bucket(bucket_name)
        blob = bucket.blob(blob_path)
        blob.upload_from_string(
            ndjson_body, content_type="application/x-ndjson"
        )

        new_poll_time = datetime.now(timezone.utc).strftime(
            "%Y-%m-%dT%H:%M:%S.000Z"
        )
        _save_state(bucket_name, state_key, new_poll_time)

        msg = (
            f"Wrote {len(objects)} STIX objects to "
            f"gs://{bucket_name}/{blob_path}"
        )
        logger.info(msg)
        return msg, 200

    except Exception as e:
        logger.error(f"Error collecting ESET TI: {e}")
        raise

  9. 배포를 클릭합니다.

  10. 함수가 배포될 때까지 기다립니다. 배포가 완료되면 상태가 녹색 체크표시로 바뀝니다.

환경 변수 구성

  1. 함수가 배포되면 Cloud Run Functions > eset-ti-collector로 이동합니다.
  2. 수정 및 새 버전 배포를 클릭합니다.
  3. 변수 및 보안 비밀 탭을 클릭합니다 (또는 1세대인 경우 런타임, 빌드, 연결, 보안 설정을 펼침).

  4. 다음 환경 변수를 추가합니다.

    키 예시 값
    GCS_BUCKET eset-ti-logs
    GCS_PREFIX eset-ti
    STATE_KEY eset-ti/state.json
    TAXII_USERNAME ETI 포털의 TAXII 사용자 이름
    TAXII_PASSWORD ETI 포털의 TAXII 비밀번호
    COLLECTION_ID 0abb06690b0b47e49cd7794396b76b20
    MAX_RECORDS 10000
    LOOKBACK_HOURS 48

  5. 배포를 클릭합니다.

Cloud Scheduler 작업 만들기

Cloud Scheduler는 일정에 따라 Pub/Sub 주제에 메시지를 게시하여 Cloud Run 함수가 ESET 위협 인텔리전스에서 새 STIX 객체를 폴링하도록 트리거합니다.

  1. Google Cloud 콘솔에서 Cloud Scheduler로 이동합니다.
  2. 작업 만들기를 클릭합니다.

  3. 다음 구성 세부정보를 제공합니다.

    설정 값
    이름 eset-ti-poll
    리전 함수와 동일한 리전을 선택합니다.
    주파수 0 */1 * * * (매시간)
    시간대 시간대를 선택합니다 (예: UTC).

  4. 계속을 클릭합니다.

  5. 실행 구성 섹션에서 다음을 수행합니다.

    • 대상 유형: Pub/Sub를 선택합니다.
    • 주제: eset-ti-trigger를 선택합니다.
    • 메시지 본문: {"poll": true}를 입력합니다.

  6. 만들기를 클릭합니다.

Cloud Run 함수 확인

  1. Cloud Scheduler에서 eset-ti-poll 작업을 찾습니다.
  2. 강제 실행을 클릭하여 즉시 실행을 트리거합니다.
  3. Cloud Run 함수 > eset-ti-collector > 로그로 이동합니다.

  4. 다음과 같은 로그 항목을 확인하여 함수가 성공적으로 실행되었는지 확인합니다.

    Fetched 250 objects (total: 250)
Wrote 250 STIX objects to gs://eset-ti-logs/eset-ti/eset_ti_0abb06690b0b47e49cd7794396b76b20_20250115_103000.json

  5. Cloud Storage > 버킷 > eset-ti-logs로 이동합니다.

  6. eset-ti/ 접두사로 이동합니다.

  7. NDJSON 파일이 STIX 객체와 함께 생성되고 있는지 확인합니다.

Google SecOps 서비스 계정을 가져오고 피드 구성

Google SecOps는 고유한 서비스 계정을 사용하여 GCS 버킷에서 데이터를 읽습니다. 이 서비스 계정에 버킷에 대한 액세스 권한을 부여해야 합니다.

서비스 계정 이메일 가져오기

  1. SIEM 설정> 피드로 이동합니다.
  2. 새 피드 추가를 클릭합니다.
  3. 단일 피드 구성을 클릭합니다.
  4. 피드 이름 필드에 피드 이름을 입력합니다(예: ESET Threat Intelligence - Botnet).
  5. 소스 유형으로 Google Cloud Storage V2를 선택합니다.
  6. 로그 유형으로 ESET 위협 인텔리전스를 선택합니다.

  7. 서비스 계정 가져오기를 클릭합니다. 고유한 서비스 계정 이메일이 표시됩니다. 예를 들면 다음과 같습니다.

    chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com

  8. 다음 단계에서 사용할 수 있도록 이 이메일 주소를 복사합니다.

  9. 다음을 클릭합니다.

  10. 다음 입력 매개변수의 값을 지정합니다.

    • 스토리지 버킷 URL: 다음 접두사 경로를 사용하여 GCS 버킷 URI를 입력합니다.

      gs://eset-ti-logs/eset-ti/

    • 소스 삭제 옵션: 환경설정에 따라 삭제 옵션을 선택합니다.

      • 삭제 안함: 전송 후 파일을 삭제하지 않습니다 (테스트에 권장).
      • 전송된 파일 삭제: 전송이 완료되면 파일을 삭제합니다.

      • 전송된 파일 및 빈 디렉터리 삭제: 전송이 완료되면 파일과 빈 디렉터리를 삭제합니다.

    • 최대 파일 기간: 지난 일수 동안 수정된 파일을 포함합니다 (기본값은 180일).

    • 애셋 네임스페이스: 애셋 네임스페이스

    • 수집 라벨: 이 피드의 이벤트에 적용할 라벨입니다 (예: ESET_IOC).

  11. 다음을 클릭합니다.

  12. 확정 화면에서 새 피드 구성을 검토한 다음 제출을 클릭합니다.

Google SecOps 서비스 계정에 IAM 권한 부여

Google SecOps 서비스 계정에는 Google Cloud Storage 버킷에 대한 스토리지 객체 뷰어 역할이 필요합니다.

  1. Cloud Storage> 버킷으로 이동합니다.
  2. 버킷 이름 (예: eset-ti-logs)을 클릭합니다.
  3. 권한 탭으로 이동합니다.
  4. 액세스 권한 부여를 클릭합니다.
  5. 다음 구성 세부정보를 제공합니다.
    • 주 구성원 추가: Google SecOps 서비스 계정 이메일을 붙여넣습니다.
    • 역할 할당: 스토리지 객체 뷰어 선택

  6. 저장을 클릭합니다.

UDM 매핑 테이블

로그 필드 UDM 매핑 논리
when metadata.event_timestamp 이벤트가 발생한 시점의 타임스탬프
metadata.event_type 이벤트 유형 (예: USER_LOGIN, NETWORK_CONNECTION)
messageid metadata.id 이벤트의 고유 식별자
프로토콜 network.ip_protocol IP 프로토콜 (예: TCP, UDP)
deviceName principal.hostname 소스 호스트 이름
srcAddr principal.ip 연결의 소스 IP 주소
srcPort principal.port 소스 포트 번호
action security_result.action 보안 제품에서 취한 조치 (예: 허용, 차단)
dstAddr target.ip 대상 IP 주소
dstPort target.port 대상 포트 번호
metadata.product_name 제품 이름
metadata.vendor_name 공급업체/회사 이름

도움이 더 필요하신가요? 커뮤니티 회원 및 Google SecOps 전문가에게 문의하여 답변을 받으세요.