CyberArk EPM 로그 수집

다음에서 지원:

이 문서에서는 AWS S3를 사용하여 CyberArk EPM 로그를 Google Security Operations에 수집하는 방법을 설명합니다. 파서는 CyberArk EPM 로그 데이터를 통합 데이터 모델 (UDM)로 변환합니다. 로그의 각 이벤트를 반복하고, 관련 필드를 해당 UDM 필드에 매핑하고, exposedUsers와 같은 특정 데이터 구조를 처리하고, 정적 공급업체 및 제품 정보로 출력을 보강합니다.

시작하기 전에

  • Google Security Operations 인스턴스가 있는지 확인합니다.
  • AWS에 대한 권한 액세스 권한이 있는지 확인합니다.
  • EPM 서버 관리 콘솔에 대한 권한 있는 액세스 권한이 있는지 확인합니다.

Google SecOps 수집을 위한 AWS IAM 구성

  1. 이 사용자 가이드(IAM 사용자 만들기)에 따라 사용자를 만듭니다.
  2. 생성된 사용자를 선택합니다.
  3. 보안 사용자 인증 정보 탭을 선택합니다.
  4. 액세스 키 섹션에서 액세스 키 만들기를 클릭합니다.
  5. 사용 사례로 서드 파티 서비스를 선택합니다.
  6. 다음을 클릭합니다.
  7. 선택사항: 설명 태그를 추가합니다.
  8. 액세스 키 만들기를 클릭합니다.
  9. CSV 파일 다운로드를 클릭하여 나중에 사용할 수 있도록 액세스 키와 비밀 액세스 키를 저장합니다.
  10. 완료를 클릭합니다.
  11. 권한 탭을 선택합니다.
  12. 권한 정책 섹션에서 권한 추가를 클릭합니다 .
  13. 권한 추가를 선택합니다.
  14. 정책 직접 연결을 선택합니다.
  15. AmazonS3FullAccess 정책을 검색하여 선택합니다.
  16. 다음을 클릭합니다.
  17. 권한 추가를 클릭합니다.

API 액세스를 위해 CyberArk EPM 구성

  1. 관리자로 CyberArk EPM 웹 콘솔에 로그인합니다.
  2. 관리 > 계정 관리로 이동합니다.
  3. + 사용자 추가를 클릭합니다.
  4. 다음 세부정보를 제공합니다.
    • 사용자 이름: epm_api_user
    • 비밀번호: 강력한 보안 비밀
    • 이메일/전체 이름: 선택사항
  5. 권한에서 가져온 로그의 모든 세트에 ViewOnlySetAdmin을 부여합니다.
  6. 저장을 클릭합니다.
  7. 선택사항: 세션 제한 시간 연장:
    • 관리 > 계정 구성으로 이동합니다.
    • 비활성 세션 시간 제한을 60분으로 설정합니다.
    • 저장을 클릭합니다.
  8. 정책 및 세트> 세트 선택> 속성으로 이동합니다.
  9. Set ID (GUID)를 복사하여 저장합니다. 스크립트에서 EPM_SET_ID로 사용합니다.

AWS S3 버킷 만들기

  1. AWS 관리 콘솔에 로그인합니다.
  2. AWS 콘솔 > 서비스 > S3 > 버킷 만들기로 이동합니다.
  3. 다음 구성 세부정보를 제공합니다.
    • 버킷 이름: my-cyberark-epm-logs
    • 지역: 선택 > 만들기

EC2용 IAM 역할 만들기

  1. AWS 관리 콘솔에 로그인합니다.
  2. 서비스로 이동합니다.
  3. 검색창에 IAM를 입력하고 선택합니다.
  4. IAM 대시보드에서 역할을 클릭합니다.
  5. 역할 만들기를 클릭합니다.
  6. 다음 구성 세부정보를 제공합니다.
    • 신뢰할 수 있는 엔티티: AWS 서비스 > EC2 > 다음
    • 권한 연결: AmazonS3FullAccess (또는 버킷에 대한 범위가 지정된 정책) > 다음
    • 역할 이름: EC2-S3-EPM-Writer > 역할 만들기

선택사항: EC2 수집기 VM 실행 및 구성

  1. AWS 관리 콘솔에 로그인합니다.
  2. 서비스로 이동합니다.
  3. 검색창에 EC2를 입력하고 선택합니다.
  4. EC2 대시보드에서 인스턴스를 클릭합니다.
  5. 인스턴스 실행을 클릭합니다.
  6. 다음 구성 세부정보를 제공합니다.
    • 이름: EPM-Log-Collector를 입력합니다.
    • AMI: Ubuntu Server 22.04 LTS를 선택합니다.
    • 인스턴스 유형: t3.micro (또는 그 이상)를 선택하고 다음을 클릭합니다.
    • 네트워크: 네트워크 설정이 기본 VPC로 설정되어 있는지 확인합니다.
    • IAM 역할: 메뉴에서 EC2-S3-EPM-Writer IAM 역할을 선택합니다.
    • 공개 IP 자동 할당: 사용 설정으로 설정합니다. VPN을 통해 연결하는 경우 이 기능을 사용 중지해도 됩니다.
    • 스토리지 추가: 기본 스토리지 구성 (8GiB)을 그대로 두고 다음을 클릭합니다.
    • 새 보안 그룹 만들기를 선택합니다.
    • 인바운드 규칙: 규칙 추가를 클릭합니다.
    • 유형: SSH를 선택합니다.
    • 포트: 22
    • 소스: 내 IP
    • 검토 및 실행을 클릭합니다.
    • 키 쌍을 선택하거나 만듭니다.
    • 키 쌍 다운로드를 클릭합니다.
    • 다운로드한 PEM 파일을 저장합니다. SSH를 통해 인스턴스에 연결하려면 이 파일이 필요합니다.

  7. SSH를 사용하여 가상 머신 (VM)에 연결합니다.

    chmod 400 ~/Downloads/your-key.pem
ssh -i ~/Downloads/your-key.pem ubuntu@<EC2_PUBLIC_IP>

수집기 기본 요건 설치

  1. 운영체제를 업데이트합니다.

    # Update OS
sudo apt update && sudo apt upgrade -y

# Install Python, Git
sudo apt install -y python3 python3-venv python3-pip git

# Create & activate virtualenv
python3 -m venv ~/epm-venv
source ~/epm-venv/bin/activate

# Install libraries
pip install requests boto3

  2. 디렉터리 및 상태 파일을 만듭니다.

    sudo mkdir -p /var/lib/epm-collector
sudo touch /var/lib/epm-collector/last_run.txt
sudo chown ubuntu:ubuntu /var/lib/epm-collector/last_run.txt

  3. 초기화합니다 (예: 1시간 전).

    echo "$(date -u -d '1 hour ago' +%Y-%m-%dT%H:%M:%SZ)" > /var/lib/epm-collector/last_run.txt

수집기 스크립트 배포

  1. 프로젝트 폴더를 만듭니다.

    mkdir ~/epm-collector && cd ~/epm-collector

  2. 환경 변수를 설정합니다 (예: ~/.bashrc).

    export EPM_URL="https://epm.mycompany.com"
export EPM_USER="epm_api_user"
export EPM_PASS="YourPasswordHere"
export EPM_SET_ID="xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx"
export S3_BUCKET="my-cyberark-epm-logs"
export S3_PREFIX="epm/"

  3. collector.py를 만들고 다음을 붙여넣습니다.

    #!/usr/bin/env python3
import os
import sys
import json
import boto3
import requests
from datetime import datetime, timezone, timedelta

# ── LOAD CONFIG FROM ENV ───────────────────────────────────────────────────────
def must_env(var):
    v = os.getenv(var)
    if not v:
        print(f"ERROR: environment variable {var} is required", file=sys.stderr)
        sys.exit(1)
    return v

EPM_URL    = must_env("EPM_URL")        # for example, https://epm.mycompany.com
USERNAME   = must_env("EPM_USER")       # API username
PASSWORD   = must_env("EPM_PASS")       # API password
SET_ID     = must_env("EPM_SET_ID")     # GUID of the Set to pull
S3_BUCKET  = must_env("S3_BUCKET")      # for example, my-cyberark-epm-logs
S3_PREFIX  = os.getenv("S3_PREFIX", "") # optional, for example "epm/"
STATE_FILE = os.getenv("STATE_FILE", "/var/lib/epm-collector/last_run.txt")
PAGE_SIZE  = int(os.getenv("PAGE_SIZE", "100"))
# ── END CONFIG ────────────────────────────────────────────────────────────────

def read_last_run():
    try:
        ts = open(STATE_FILE).read().strip()
        return datetime.fromisoformat(ts.replace("Z","+00:00"))
    except:
        # default to 1 hour ago
        return datetime.now(timezone.utc) - timedelta(hours=1)

def write_last_run(dt):
    with open(STATE_FILE, "w") as f:
        f.write(dt.strftime("%Y-%m-%dT%H:%M:%SZ"))

def logon():
    r = requests.post(
        f"{EPM_URL}/REST/EPMService.svc/Logon",
        json={"username": USERNAME, "password": PASSWORD},
        headers={"Content-Type": "application/json"}
    )
    r.raise_for_status()
    return r.json().get("SessionToken")

def logoff(token):
    requests.post(
        f"{EPM_URL}/REST/EPMService.svc/Logoff",
        headers={"Authorization": f"Bearer {token}"}
    )

def fetch_raw_events(token, start, end):
    headers = {"Authorization": f"Bearer {token}"}
    page = 1
    while True:
        params = {
            "setId":     SET_ID,
            "startDate": start,
            "endDate":   end,
            "pageSize":  PAGE_SIZE,
            "pageNumber": page
        }
        resp = requests.get(
            f"{EPM_URL}/REST/EPMService.svc/GetRawEvents",
            headers=headers, params=params
        )
        resp.raise_for_status()
        events = resp.json().get("RawEvents", [])
        if not events:
            break
        yield from events
        page += 1

def upload_to_s3(obj, key):
    boto3.client("s3").put_object(
        Bucket=S3_BUCKET,
        Key=key,
        Body=json.dumps(obj).encode("utf-8")
    )

def main():
    # determine time window
    start_dt = read_last_run()
    end_dt   = datetime.now(timezone.utc)
    START = start_dt.strftime("%Y-%m-%dT%H:%M:%SZ")
    END   = end_dt.strftime("%Y-%m-%dT%H:%M:%SZ")

    token = logon()
    try:
        for idx, raw_evt in enumerate(fetch_raw_events(token, START, END), start=1):
            key = f"{S3_PREFIX}{end_dt.strftime('%Y/%m/%d')}/raw_{int(end_dt.timestamp())}_{idx}.json"
            upload_to_s3(raw_evt, key)
            print(f"Uploaded raw event to {key}")
    finally:
        logoff(token)

    # persist for next run
    write_last_run(end_dt)

if __name__ == "__main__":
    main()

  4. 스크립트를 실행 가능하게 만듭니다.

    chmod +x collector.py

Cron으로 자동화

  1. crontab을 엽니다.

    crontab -e

  2. 일일 작업을 추가합니다.

    0 0 * * * cd ~/epm-collector && source ~/epm-venv/bin/activate && python collector.py >> ~/epm-collector/epm.log 2>&1

Cyberark EPM 로그를 수집하도록 Google SecOps에서 피드 구성

  1. SIEM 설정> 피드로 이동합니다.
  2. 새로 추가를 클릭합니다.
  3. 피드 이름 필드에 피드 이름을 입력합니다 (예: Cyberark EPM Logs).
  4. 소스 유형으로 Amazon S3를 선택합니다.
  5. 로그 유형으로 Cyberark EPM을 선택합니다.
  6. 다음을 클릭합니다.

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

    • 리전: Amazon S3 버킷이 있는 리전입니다.
    • S3 URI: 버킷 URI (형식은 s3://your-log-bucket-name/이어야 함). 다음을 대체합니다.
      • your-log-bucket-name: 버킷의 이름입니다.
    • URI is a: Directory 또는 Directory which includes subdirectories를 선택합니다.
    • 소스 삭제 옵션: 환경설정에 따라 삭제 옵션을 선택합니다.
    • 액세스 키 ID: s3 버킷에 액세스할 수 있는 사용자 액세스 키입니다.
    • 보안 비밀 액세스 키: s3 버킷에 액세스할 수 있는 사용자 보안 비밀 키입니다.
    • 애셋 네임스페이스: 애셋 네임스페이스입니다.
    • 수집 라벨: 이 피드의 이벤트에 적용할 라벨입니다.

  8. 다음을 클릭합니다.

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

UDM 매핑 테이블

로그 필드 UDM 매핑 논리
agentId principal.asset.asset_id 'agentId:'를 agentId 필드의 값과 연결합니다.
computerName principal.hostname computerName 필드를 직접 매핑합니다.
displayName metadata.description displayName 필드를 직접 매핑합니다.
eventType metadata.product_event_type eventType 필드를 직접 매핑합니다.
exposedUsers.[].accountName target.user.attribute.labels 키가 'accountName_[index]'이고 값이 exposedUsers.[index].accountName인 라벨을 만듭니다.
exposedUsers.[].domain target.user.attribute.labels 키가 'domain_[index]'이고 값이 exposedUsers.[index].domain인 라벨을 만듭니다.
exposedUsers.[].username target.user.attribute.labels 키가 'username_[index]'이고 값이 exposedUsers.[index].username인 라벨을 만듭니다.
filePath target.file.full_path filePath 필드를 직접 매핑합니다.
해시 target.file.sha1 해시 필드를 직접 매핑합니다.
operatingSystemType principal.platform operatingSystemType 필드가 'Windows'인 경우 'Windows'를 'WINDOWS'로 매핑합니다.
policyName security_result.rule_name policyName 필드를 직접 매핑합니다.
processCommandLine target.process.command_line processCommandLine 필드를 직접 매핑합니다.
게시자 additional.fields 게시자 필드에서 키가 'Publisher'이고 string_value가 있는 필드를 만듭니다.
sourceProcessCommandLine target.process.parent_process.command_line sourceProcessCommandLine 필드를 직접 매핑합니다.
sourceProcessHash target.process.parent_process.file.sha1 sourceProcessHash 필드를 직접 매핑합니다.
sourceProcessSigner additional.fields sourceProcessSigner 필드에서 키가 'sourceProcessSigner'이고 string_value가 있는 필드를 만듭니다.
threatProtectionAction security_result.action_details threatProtectionAction 필드를 직접 매핑합니다.
metadata.event_timestamp 이벤트 타임스탬프를 로그 항목의 create_time으로 설정합니다.
metadata.event_type 'STATUS_UPDATE'로 하드코딩됩니다.
metadata.log_type 'CYBERARK_EPM'으로 하드코딩됩니다.
metadata.product_name 'EPM'으로 하드코딩됨
metadata.vendor_name 'CYBERARK'로 하드코딩됩니다.
security_result.alert_state 'ALERTING'으로 하드코딩됩니다.
userName principal.user.userid userName 필드를 직접 매핑합니다.

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