Tines 감사 로그 수집
이 문서에서는 Google Cloud Storage를 사용하여 Tines 감사 로그를 Google Security Operations에 수집하는 방법을 설명합니다.
Tines는 보안팀이 워크플로를 구축하고 보안 운영을 자동화할 수 있는 노코드 자동화 플랫폼입니다. Tines 감사 로그는 Tines 플랫폼 내의 사용자 작업, 구성 변경사항, 시스템 이벤트를 파악할 수 있도록 지원합니다.
시작하기 전에
다음 기본 요건이 충족되었는지 확인합니다.
- Google SecOps 인스턴스
- Cloud Storage API가 사용 설정된 GCP 프로젝트
- GCS 버킷을 만들고 관리할 수 있는 권한
- GCS 버킷의 IAM 정책을 관리할 수 있는 권한
- Cloud Run 서비스, Pub/Sub 주제, Cloud Scheduler 작업을 만들 수 있는 권한
- Tines에 대한 액세스 권한 관리
Tines URL 가져오기
- 브라우저에서 테넌트의 Tines UI를 엽니다.
- 주소 표시줄에서 도메인을 복사합니다. 이 도메인은
TINES_BASE_URL로 사용됩니다. - 형식:
https://<tenant-domain>(예:https://<tenant-domain>.tines.com)
Tines 서비스 API 키 (권장) 또는 개인 API 키 만들기
나중에 사용할 값을 저장합니다.
TINES_BASE_URL(예:https://<domain>.tines.com)TINES_API_KEY- 다음 단계에서 만드는 토큰
옵션 1 - 서비스 API 키 (권장)
- 탐색 메뉴 > API 키로 이동합니다.
- + 새 키를 클릭합니다.
- 서비스 API 키를 선택합니다.
- 설명이 포함된 이름 (예:
SecOps Audit Logs)을 입력합니다. - 만들기를 클릭합니다.
- 생성된 토큰을 즉시 복사하고 안전하게 저장합니다. 이 토큰은
TINES_API_KEY로 사용됩니다.
옵션 2 - 개인 API 키 (서비스 키를 사용할 수 없는 경우)
- 탐색 메뉴 > API 키로 이동합니다.
- + 새 키를 클릭합니다.
- 개인 API 키를 선택합니다.
- 설명을 포함한 이름을 입력합니다.
- 만들기를 클릭합니다.
생성된 토큰을 복사하여 안전하게 저장합니다.
감사 로그 읽기 권한 부여
- 테넌트 소유자로 로그인하거나 테넌트 소유자에게 로그인을 요청합니다.
- 설정 > 관리 > 사용자 관리로 이동하거나 왼쪽 상단 메뉴에서 팀 이름을 클릭하고 사용자를 선택합니다.
- 서비스 API 키와 연결된 서비스 계정 사용자를 찾습니다 (API 키와 이름이 동일함). 개인 API 키를 사용하는 경우 대신 자신의 사용자 계정을 찾으세요.
- 사용자를 클릭하여 프로필을 엽니다.
- 테넌트 권한 섹션에서 AUDIT_LOG_READ를 사용 설정합니다.
- 저장을 클릭합니다.
권한 확인
계정에 필요한 권한이 있는지 확인하려면 다음 단계를 따르세요.
- Tines에 로그인합니다.
- 설정 > 모니터링 > 감사 로그로 이동합니다.
- 감사 로그 항목이 표시되면 필요한 권한이 있는 것입니다.
- 이 옵션이 표시되지 않으면 테넌트 소유자에게 문의하여 AUDIT_LOG_READ 권한을 부여받으세요.
API 액세스 테스트
통합을 진행하기 전에 사용자 인증 정보를 테스트하세요.
# Replace with your actual credentials TINES_BASE_URL="https://<tenant-domain>.tines.com" TINES_API_KEY="<your-api-key>" # Test API access curl -X GET "${TINES_BASE_URL}/api/v1/audit_logs?per_page=1" \ -H "Authorization: Bearer ${TINES_API_KEY}" \ -H "Content-Type: application/json"
감사 로그 항목이 포함된 JSON 응답이 표시됩니다.
Google Cloud Storage 버킷 만들기
- Google Cloud Console로 이동합니다.
- 프로젝트를 선택하거나 새 프로젝트를 만듭니다.
- 탐색 메뉴에서 Cloud Storage> 버킷으로 이동합니다.
- 버킷 만들기를 클릭합니다.
다음 구성 세부정보를 제공합니다.
설정 값 버킷 이름 지정 전역적으로 고유한 이름 (예: tines-audit-logs)을 입력합니다.위치 유형 필요에 따라 선택 (리전, 이중 리전, 멀티 리전) 위치 위치를 선택합니다 (예: us-central1).스토리지 클래스 Standard (자주 액세스하는 로그에 권장) 액세스 제어 균일 (권장) 보호 도구 선택사항: 객체 버전 관리 또는 보관 정책 사용 설정 만들기를 클릭합니다.
Cloud Run 함수의 서비스 계정 만들기
Cloud Run 함수에는 GCS 버킷에 쓸 수 있고 Pub/Sub에서 호출할 수 있는 권한이 있는 서비스 계정이 필요합니다.
서비스 계정 만들기
- GCP 콘솔에서 IAM 및 관리자 > 서비스 계정으로 이동합니다.
- 서비스 계정 만들기를 클릭합니다.
- 다음 구성 세부정보를 제공합니다.
- 서비스 계정 이름:
tines-audit-to-gcs-sa을 입력합니다. - 서비스 계정 설명:
Service account for Cloud Run function to collect Tines Audit Logs을 입력합니다.
- 서비스 계정 이름:
- 만들고 계속하기를 클릭합니다.
- 이 서비스 계정에 프로젝트에 대한 액세스 권한 부여 섹션에서 다음 역할을 추가합니다.
- 역할 선택을 클릭합니다.
- 스토리지 객체 관리자를 검색하여 선택합니다.
- + 다른 역할 추가를 클릭합니다.
- Cloud Run 호출자를 검색하여 선택합니다.
- + 다른 역할 추가를 클릭합니다.
- Cloud Functions 호출자를 검색하여 선택합니다.
- 계속을 클릭합니다.
- 완료를 클릭합니다.
이러한 역할은 다음 작업에 필요합니다.
- 스토리지 객체 관리자: GCS 버킷에 로그를 쓰고 상태 파일을 관리합니다.
- Cloud Run 호출자: Pub/Sub가 함수를 호출하도록 허용
- Cloud Functions 호출자: 함수 호출 허용
GCS 버킷에 대한 IAM 권한 부여
GCS 버킷에 대한 쓰기 권한을 서비스 계정에 부여합니다.
- Cloud Storage> 버킷으로 이동합니다.
- 버킷 이름을 클릭합니다.
- 권한 탭으로 이동합니다.
- 액세스 권한 부여를 클릭합니다.
- 다음 구성 세부정보를 제공합니다.
- 주 구성원 추가: 서비스 계정 이메일 (예:
tines-audit-to-gcs-sa@PROJECT_ID.iam.gserviceaccount.com)을 입력합니다. - 역할 할당: 스토리지 객체 관리자를 선택합니다.
- 주 구성원 추가: 서비스 계정 이메일 (예:
- 저장을 클릭합니다.
게시/구독 주제 만들기
Cloud Scheduler가 게시하고 Cloud Run 함수가 구독할 Pub/Sub 주제를 만듭니다.
- GCP Console에서 Pub/Sub > 주제로 이동합니다.
- 주제 만들기를 클릭합니다.
- 다음 구성 세부정보를 제공합니다.
- 주제 ID:
tines-audit-trigger를 입력합니다. - 다른 설정은 기본값으로 둡니다.
- 주제 ID:
- 만들기를 클릭합니다.
로그를 수집하는 Cloud Run 함수 만들기
Cloud Run 함수는 Cloud Scheduler의 Pub/Sub 메시지에 의해 트리거되어 Tines API에서 로그를 가져오고 GCS에 기록합니다.
- GCP 콘솔에서 Cloud Run으로 이동합니다.
- 서비스 만들기를 클릭합니다.
- 함수를 선택합니다 (인라인 편집기를 사용하여 함수 만들기).
구성 섹션에서 다음 구성 세부정보를 제공합니다.
설정 값 서비스 이름 tines-audit-to-gcs리전 GCS 버킷과 일치하는 리전을 선택합니다 (예: us-central1).런타임 Python 3.12 이상 선택 트리거 (선택사항) 섹션에서 다음을 수행합니다.
- + 트리거 추가를 클릭합니다.
- Cloud Pub/Sub를 선택합니다.
- Cloud Pub/Sub 주제 선택에서 Pub/Sub 주제 (
tines-audit-trigger)를 선택합니다. - 저장을 클릭합니다.
인증 섹션에서 다음을 구성합니다.
- 인증 필요를 선택합니다.
- ID 및 액세스 관리 (IAM)를 확인합니다.
아래로 스크롤하고 컨테이너, 네트워킹, 보안을 펼칩니다.
보안 탭으로 이동합니다.
- 서비스 계정: 서비스 계정 (
tines-audit-to-gcs-sa)을 선택합니다.
- 서비스 계정: 서비스 계정 (
컨테이너 탭으로 이동합니다.
- 변수 및 보안 비밀을 클릭합니다.
- 각 환경 변수에 대해 + 변수 추가를 클릭합니다.
변수 이름 예시 값 설명 GCS_BUCKETtines-audit-logsGCS 버킷 이름 GCS_PREFIXtines/audit/로그 파일의 접두사 STATE_KEYtines/audit/state.json상태 파일 경로 TINES_BASE_URLhttps://your-tenant.tines.comAPI 기본 URL TINES_API_KEYyour-tines-api-keyAPI 키 LOOKBACK_SECONDS3600초기 전환 확인 기간 PAGE_SIZE500페이지당 레코드 수 MAX_PAGES20실행당 최대 페이지 수 HTTP_TIMEOUT60HTTP 요청 시간초과 HTTP_RETRIES3HTTP 재시도 횟수 변수 및 보안 비밀 섹션에서 요청까지 아래로 스크롤합니다.
- 요청 제한 시간:
600초 (10분)를 입력합니다.
- 요청 제한 시간:
설정 탭으로 이동합니다.
- 리소스 섹션에서 다음을 수행합니다.
- 메모리: 512MiB 이상을 선택합니다.
- CPU: 1을 선택합니다.
- 리소스 섹션에서 다음을 수행합니다.
버전 확장 섹션에서 다음을 수행합니다.
- 최소 인스턴스 수:
0를 입력합니다. - 최대 인스턴스 수:
100을 입력합니다 (또는 예상 부하에 따라 조정).
- 최소 인스턴스 수:
만들기를 클릭합니다.
서비스가 생성될 때까지 기다립니다 (1~2분).
서비스가 생성되면 인라인 코드 편집기가 자동으로 열립니다.
함수 코드 추가
- 함수 진입점에 main을 입력합니다.
인라인 코드 편집기에서 다음 두 파일을 만듭니다.
- 첫 번째 파일: main.py:
import functions_framework from google.cloud import storage import json import os import urllib3 from datetime import datetime, timezone, timedelta import time # Initialize HTTP client with timeouts http = urllib3.PoolManager( timeout=urllib3.Timeout(connect=5.0, read=30.0), retries=False, ) # Initialize Storage client storage_client = storage.Client() # Environment variables GCS_BUCKET = os.environ.get('GCS_BUCKET') GCS_PREFIX = os.environ.get('GCS_PREFIX', 'tines/audit/') STATE_KEY = os.environ.get('STATE_KEY', 'tines/audit/state.json') TINES_BASE_URL = os.environ.get('TINES_BASE_URL') TINES_API_KEY = os.environ.get('TINES_API_KEY') LOOKBACK_SECONDS = int(os.environ.get('LOOKBACK_SECONDS', '3600')) PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '500')) MAX_PAGES = int(os.environ.get('MAX_PAGES', '20')) HTTP_TIMEOUT = int(os.environ.get('HTTP_TIMEOUT', '60')) HTTP_RETRIES = int(os.environ.get('HTTP_RETRIES', '3')) def parse_datetime(value: str) -> datetime: """Parse ISO datetime string to datetime object.""" if value.endswith("Z"): value = value[:-1] + "+00:00" return datetime.fromisoformat(value) @functions_framework.cloud_event def main(cloud_event): """ Cloud Run function triggered by Pub/Sub to fetch Tines Audit Logs and write to GCS. Args: cloud_event: CloudEvent object containing Pub/Sub message """ if not all([GCS_BUCKET, TINES_BASE_URL, TINES_API_KEY]): print('Error: Missing required environment variables') return try: # Get GCS bucket bucket = storage_client.bucket(GCS_BUCKET) # Load state state = load_state(bucket, STATE_KEY) # Determine time window now = datetime.now(timezone.utc) last_time = None if isinstance(state, dict) and state.get("last_event_time"): try: last_time = parse_datetime(state["last_event_time"]) # Overlap by 2 minutes to catch any delayed events last_time = last_time - timedelta(minutes=2) except Exception as e: print(f"Warning: Could not parse last_event_time: {e}") if last_time is None: last_time = now - timedelta(seconds=LOOKBACK_SECONDS) print(f"Fetching logs from {last_time.isoformat()} to {now.isoformat()}") # Fetch logs records, newest_event_time = fetch_logs( api_base=TINES_BASE_URL, api_key=TINES_API_KEY, start_time=last_time, page_size=PAGE_SIZE, max_pages=MAX_PAGES, ) if not records: print("No new log records found.") save_state(bucket, STATE_KEY, now.isoformat()) return # Write to GCS as NDJSON timestamp = now.strftime('%Y%m%d_%H%M%S') object_key = f"{GCS_PREFIX}/logs_{timestamp}.ndjson" blob = bucket.blob(object_key) ndjson = '\n'.join([json.dumps(record, ensure_ascii=False) for record in records]) + '\n' blob.upload_from_string(ndjson, content_type='application/x-ndjson') print(f"Wrote {len(records)} records to gs://{GCS_BUCKET}/{object_key}") # Update state with newest event time if newest_event_time: save_state(bucket, STATE_KEY, newest_event_time) else: save_state(bucket, STATE_KEY, now.isoformat()) print(f"Successfully processed {len(records)} records") except Exception as e: print(f'Error processing logs: {str(e)}') raise def load_state(bucket, key): """Load state from GCS.""" try: blob = bucket.blob(key) if blob.exists(): state_data = blob.download_as_text() return json.loads(state_data) except Exception as e: print(f"Warning: Could not load state: {e}") return {} def save_state(bucket, key, last_event_time_iso: str): """Save the last event timestamp to GCS state file.""" try: state = {'last_event_time': last_event_time_iso} blob = bucket.blob(key) blob.upload_from_string( json.dumps(state, indent=2), content_type='application/json' ) print(f"Saved state: last_event_time={last_event_time_iso}") except Exception as e: print(f"Warning: Could not save state: {e}") def fetch_logs(api_base: str, api_key: str, start_time: datetime, page_size: int, max_pages: int): """ Fetch logs from Tines API with pagination and rate limiting. Args: api_base: API base URL api_key: Tines API key start_time: Start time for log query page_size: Number of records per page max_pages: Maximum pages to fetch Returns: Tuple of (records list, newest_event_time ISO string) """ base_url = api_base.rstrip('/') endpoint = f"{base_url}/api/v1/audit_logs" headers = { 'Authorization': f'Bearer {api_key}', 'Accept': 'application/json', 'Content-Type': 'application/json', 'User-Agent': 'GoogleSecOps-TinesCollector/1.0' } records = [] newest_time = None page_num = 1 backoff = 1.0 while page_num <= max_pages: # Build URL with query parameters params = { 'page': page_num, 'per_page': page_size } param_str = '&'.join([f"{k}={v}" for k, v in params.items()]) url = f"{endpoint}?{param_str}" try: response = http.request('GET', url, headers=headers, timeout=HTTP_TIMEOUT) # Handle rate limiting with exponential backoff if response.status == 429: retry_after = int(response.headers.get('Retry-After', str(int(backoff)))) print(f"Rate limited (429). Retrying after {retry_after}s...") time.sleep(retry_after) backoff = min(backoff * 2, 30.0) continue backoff = 1.0 if response.status != 200: print(f"HTTP Error: {response.status}") response_text = response.data.decode('utf-8') print(f"Response body: {response_text}") return [], None data = json.loads(response.data.decode('utf-8')) # Extract results page_results = [] if isinstance(data, dict): page_results = data.get('audit_logs', []) elif isinstance(data, list): page_results = data if not page_results: print(f"No more results (empty page)") break # Filter by start_time filtered_results = [] for event in page_results: try: event_time = event.get('created_at') if event_time: event_dt = parse_datetime(event_time) if event_dt >= start_time: filtered_results.append(event) # Track newest event time if newest_time is None or event_dt > parse_datetime(newest_time): newest_time = event_time except Exception as e: print(f"Warning: Could not parse event time: {e}") filtered_results.append(event) print(f"Page {page_num}: Retrieved {len(page_results)} events, {len(filtered_results)} after filtering") records.extend(filtered_results) # Check for more results if isinstance(data, dict): meta = data.get('meta', {}) next_page = meta.get('next_page_number') if not next_page: print("No more pages (no next_page_number)") break page_num = next_page else: # If response is a list, check if we got fewer results than requested if len(page_results) < page_size: print(f"Reached last page (size={len(page_results)} < limit={page_size})") break page_num += 1 except Exception as e: print(f"Error fetching logs: {e}") if page_num <= HTTP_RETRIES: print(f"Retrying... (attempt {page_num}/{HTTP_RETRIES})") time.sleep(backoff) backoff = min(backoff * 2, 30.0) continue return [], None print(f"Retrieved {len(records)} total records from {page_num} pages") return records, newest_time- 두 번째 파일: requirements.txt:
functions-framework==3.* google-cloud-storage==2.* urllib3>=2.0.0배포를 클릭하여 함수를 저장하고 배포합니다.
배포가 완료될 때까지 기다립니다 (2~3분).
Cloud Scheduler 작업 만들기
Cloud Scheduler는 일정 간격으로 Pub/Sub 주제에 메시지를 게시하여 Cloud Run 함수를 트리거합니다.
- GCP Console에서 Cloud Scheduler로 이동합니다.
- 작업 만들기를 클릭합니다.
다음 구성 세부정보를 제공합니다.
설정 값 이름 tines-audit-hourly리전 Cloud Run 함수와 동일한 리전 선택 주파수 0 * * * *(매시간 정각)시간대 시간대 선택 (UTC 권장) 타겟 유형 Pub/Sub 주제 Pub/Sub 주제 ( tines-audit-trigger)를 선택합니다.메일 본문 {}(빈 JSON 객체)만들기를 클릭합니다.
일정 빈도 옵션
로그 볼륨 및 지연 시간 요구사항에 따라 빈도를 선택합니다.
빈도 크론 표현식 사용 사례 5분마다 */5 * * * *대용량, 저지연 15분마다 */15 * * * *검색량 보통 1시간마다 0 * * * *일반(권장) 6시간마다 0 */6 * * *양이 적은 일괄 처리 매일 0 0 * * *이전 데이터 수집
통합 테스트
- Cloud Scheduler 콘솔에서 작업을 찾습니다 (
tines-audit-hourly). - 강제 실행을 클릭하여 작업을 수동으로 트리거합니다.
- 몇 초 동안 기다립니다.
- Cloud Run > 서비스로 이동합니다.
- 함수 이름 (
tines-audit-to-gcs)을 클릭합니다. - 로그 탭을 클릭합니다.
함수가 성공적으로 실행되었는지 확인합니다. 다음을 확인하세요.
Fetching logs from YYYY-MM-DDTHH:MM:SS+00:00 to YYYY-MM-DDTHH:MM:SS+00:00 Page 1: Retrieved X events Wrote X records to gs://bucket-name/tines/audit/logs_YYYYMMDD_HHMMSS.ndjson Successfully processed X recordsCloud Storage> 버킷으로 이동합니다.
버킷 이름 (
tines-audit-logs)을 클릭합니다.tines/audit/폴더로 이동합니다.현재 타임스탬프를 사용하여 새
.ndjson파일이 생성되었는지 확인합니다.
로그에 오류가 표시되면 다음 단계를 따르세요.
- HTTP 401: 환경 변수에서 API 사용자 인증 정보 확인
- HTTP 403: 계정에 AUDIT_LOG_READ 권한이 있는지 확인
- HTTP 429: 비율 제한 - 함수가 백오프를 사용하여 자동으로 재시도됩니다.
- 환경 변수 누락: 필수 변수가 모두 설정되었는지 확인
Google SecOps 서비스 계정 가져오기
Google SecOps는 고유한 서비스 계정을 사용하여 GCS 버킷에서 데이터를 읽습니다. 이 서비스 계정에 버킷에 대한 액세스 권한을 부여해야 합니다.
서비스 계정 이메일 가져오기
- SIEM 설정> 피드로 이동합니다.
- 새 피드 추가를 클릭합니다.
- 단일 피드 구성을 클릭합니다.
- 피드 이름 필드에 피드 이름을 입력합니다(예:
Tines Audit Logs). - 소스 유형으로 Google Cloud Storage V2를 선택합니다.
- 로그 유형으로 Tines를 선택합니다.
서비스 계정 가져오기를 클릭합니다. 고유한 서비스 계정 이메일이 표시됩니다. 예를 들면 다음과 같습니다.
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com다음 단계에서 사용할 수 있도록 이 이메일 주소를 복사합니다.
Google SecOps 서비스 계정에 IAM 권한 부여
Google SecOps 서비스 계정에는 GCS 버킷에 대한 스토리지 객체 뷰어 역할이 필요합니다.
- Cloud Storage> 버킷으로 이동합니다.
- 버킷 이름 (
tines-audit-logs)을 클릭합니다. - 권한 탭으로 이동합니다.
- 액세스 권한 부여를 클릭합니다.
- 다음 구성 세부정보를 제공합니다.
- 주 구성원 추가: Google SecOps 서비스 계정 이메일을 붙여넣습니다.
- 역할 할당: 스토리지 객체 뷰어를 선택합니다.
저장을 클릭합니다.
Tines 감사 로그를 수집하도록 Google SecOps에서 피드 구성
- SIEM 설정> 피드로 이동합니다.
- 새 피드 추가를 클릭합니다.
- 단일 피드 구성을 클릭합니다.
- 피드 이름 필드에 피드 이름을 입력합니다(예:
Tines Audit Logs). - 소스 유형으로 Google Cloud Storage V2를 선택합니다.
- 로그 유형으로 Tines를 선택합니다.
- 다음을 클릭합니다.
다음 입력 매개변수의 값을 지정합니다.
스토리지 버킷 URL: 다음 접두사 경로를 사용하여 GCS 버킷 URI를 입력합니다.
gs://tines-audit-logs/tines/audit/다음과 같이 바꿉니다.
tines-audit-logs: GCS 버킷 이름입니다.tines/audit/: 로그가 저장되는 접두사/폴더 경로입니다.
소스 삭제 옵션: 환경설정에 따라 삭제 옵션을 선택합니다.
- 삭제 안함: 전송 후 파일을 삭제하지 않습니다 (테스트에 권장).
- 전송된 파일 삭제: 전송이 완료되면 파일을 삭제합니다.
전송된 파일 및 빈 디렉터리 삭제: 전송이 완료되면 파일과 빈 디렉터리를 삭제합니다.
최대 파일 기간: 지난 일수 동안 수정된 파일을 포함합니다. 기본값은 180일입니다.
애셋 네임스페이스: 애셋 네임스페이스입니다.
수집 라벨: 이 피드의 이벤트에 적용할 라벨입니다.
다음을 클릭합니다.
확정 화면에서 새 피드 구성을 검토한 다음 제출을 클릭합니다.
도움이 더 필요한가요? 커뮤니티 회원 및 Google SecOps 전문가에게 문의하여 답변을 받으세요.