Oracle Cloud Infrastructure 감사 로그 수집

다음에서 지원:

이 문서에서는 다음 방법 중 하나를 사용하여 Oracle Cloud Infrastructure (OCI) 감사 로그를 Google Security Operations에 수집하는 방법을 설명합니다.

  • 방법 1: Cloud Storage: OCI 함수가 로그를 Cloud Storage 버킷에 쓰고 Google SecOps 피드가 버킷에서 로그를 읽습니다.
  • 방법 2: 웹훅: OCI 함수가 로그를 Google SecOps 웹훅 피드로 직접 전송합니다. 이 방법은 Cloud Storage 버킷을 사용하지 않습니다.

OCI는 Cloud Storage 또는 HTTPS 엔드포인트로 로그를 직접 전송할 수 없으므로 두 방법 모두 OCI Connector Hub를 사용하여 감사 로그를 함수에 전달합니다.

Oracle Cloud Infrastructure Audit 서비스는 지원되는 모든 OCI 공개 API 엔드포인트에 대한 호출을 로그 이벤트로 자동 기록합니다. 모든 서비스는 OCI 감사를 통한 로깅을 지원합니다. OCI 감사에서 기록하는 로그 이벤트에는 OCI 콘솔, CLI, SDK, 자체 맞춤 클라이언트 또는 기타 OCI 서비스에서 수행한 API 호출이 포함됩니다.

시작하기 전에

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

  • Google SecOps 인스턴스
  • 다음을 만들고 관리할 수 있는 권한이 있는 Oracle Cloud 콘솔에 대한 액세스 권한:
    • 커넥터 허브 커넥터
    • Functions 애플리케이션 및 함수
    • IAM 정책
  • OCI 함수에 대한 인터넷 액세스 권한이 있는 VCN 서브넷
  • OCI Cloud Shell 또는 Fn CLI가 있는 로컬 환경
  • 방법 1: Cloud Storage API가 사용 설정된 Google Cloud 프로젝트, Cloud Storage 버킷, 서비스 계정, 버킷 Identity and Access Management (IAM) 정책을 만들 수 있는 권한
  • 방법 2: Google SecOps 인스턴스에 연결된 Google Cloud 프로젝트에서 API 키를 만들 수 있는 권한

방법 1: Cloud Storage

Cloud Storage 버킷 만들기

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

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

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

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

Cloud Storage로 감사 로그를 내보내도록 OCI 구성

Oracle Cloud Infrastructure는 Cloud Storage로의 기본 제공 내보내기를 지원하지 않습니다. 함수와 함께 OCI Service Connector Hub를 사용하여 감사 로그를 Cloud Storage로 전달합니다.

Cloud Storage로 로그를 전달하는 OCI 함수 만들기

  1. Oracle Cloud 콘솔에 로그인합니다.
  2. 개발자 서비스 > 함수 > 애플리케이션으로 이동합니다.
  3. 함수 애플리케이션을 만들 구획을 선택합니다.
  4. 애플리케이션 만들기를 클릭합니다.
  5. 다음 구성 세부정보를 제공합니다.
    • 이름: audit-logs-to-gcs-app를 입력합니다.
    • VCN: 가상 클라우드 네트워크를 선택합니다.
    • 서브넷: 인터넷 액세스가 있는 서브넷을 선택합니다.
  6. 만들기를 클릭합니다.
  7. 애플리케이션이 생성되면 시작하기를 클릭하고 안내에 따라 Fn CLI를 사용하여 로컬 개발 환경을 설정합니다.

  8. 로컬 머신에 새 함수 디렉터리를 만듭니다.

    mkdir oci-audit-to-gcs
cd oci-audit-to-gcs

  9. Python 함수를 초기화합니다.

    fn init --runtime python oci-audit-to-gcs
cd oci-audit-to-gcs

  10. func.py의 내용을 다음 코드로 바꿉니다.

    import io
import json
import logging
import os
from fdk import response
from google.cloud import storage
from google.oauth2 import service_account
from datetime import datetime

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger()

# Environment variables
GCS_BUCKET = os.environ.get('GCS_BUCKET')
GCS_PREFIX = os.environ.get('GCS_PREFIX', 'oci-audit-logs')
GCS_CREDENTIALS_JSON = os.environ.get('GCS_CREDENTIALS_JSON')

def handler(ctx, data: io.BytesIO = None):
    """
    Oracle Cloud Infrastructure Function to forward Audit logs to GCS.

    Args:
        ctx: Function context
        data: Input data containing Audit log events
    """

    if not all([GCS_BUCKET, GCS_CREDENTIALS_JSON]):
        logger.error('Missing required environment variables: GCS_BUCKET or GCS_CREDENTIALS_JSON')
        return response.Response(
            ctx, response_data=json.dumps({"error": "Missing configuration"}),
            headers={"Content-Type": "application/json"}
        )

    try:
        # Parse input data
        body = json.loads(data.getvalue())
        logger.info(f"Received event: {json.dumps(body)}")

        # Extract log entries
        log_entries = []
        if isinstance(body, list):
            log_entries = body
        elif isinstance(body, dict):
            # Service Connector Hub sends data in specific format
            if 'data' in body:
                log_entries = [body['data']] if isinstance(body['data'], dict) else body['data']
            else:
                log_entries = [body]

        if not log_entries:
            logger.info("No log entries to process")
            return response.Response(
                ctx, response_data=json.dumps({"status": "no_logs"}),
                headers={"Content-Type": "application/json"}
            )

        # Initialize GCS client with service account credentials
        credentials_dict = json.loads(GCS_CREDENTIALS_JSON)
        credentials = service_account.Credentials.from_service_account_info(credentials_dict)
        storage_client = storage.Client(credentials=credentials, project=credentials_dict.get('project_id'))
        bucket = storage_client.bucket(GCS_BUCKET)

        # Write logs to GCS as NDJSON
        timestamp = datetime.utcnow().strftime('%Y%m%d_%H%M%S_%f')
        object_key = f"{GCS_PREFIX}/logs_{timestamp}.ndjson"
        blob = bucket.blob(object_key)

        ndjson = '\n'.join([json.dumps(entry, ensure_ascii=False) for entry in log_entries]) + '\n'
        blob.upload_from_string(ndjson, content_type='application/x-ndjson')

        logger.info(f"Wrote {len(log_entries)} records to gs://{GCS_BUCKET}/{object_key}")

        return response.Response(
            ctx, response_data=json.dumps({"status": "success", "records": len(log_entries)}),
            headers={"Content-Type": "application/json"}
        )

    except Exception as e:
        logger.error(f'Error processing logs: {str(e)}')
        return response.Response(
            ctx, response_data=json.dumps({"error": str(e)}),
            headers={"Content-Type": "application/json"},
            status_code=500
        )

  11. 다음 종속 항목으로 requirements.txt를 업데이트합니다.

    fdk>=0.1.0
google-cloud-storage>=2.0.0
google-auth>=2.0.0

  12. OCI에 함수를 배포합니다.

    fn -v deploy --app audit-logs-to-gcs-app

  13. 배포가 완료되면 함수 OCID를 기록해 둡니다. 다음 단계에서 사용합니다.

함수 환경 변수 구성

  1. Oracle Cloud 콘솔에서 개발자 서비스 > 함수 > 애플리케이션으로 이동합니다.
  2. 애플리케이션 (audit-logs-to-gcs-app)을 클릭합니다.
  3. 함수 이름 (oci-audit-to-gcs)을 클릭합니다.
  4. 구성을 클릭합니다.

  5. 다음 구성 변수를 추가합니다.

    키 값
    GCS_BUCKET Cloud Storage 버킷 이름 (예: oci-audit-logs-gcs)
    GCS_PREFIX 로그 파일의 프리픽스 (예: oci-audit-logs)
    GCS_CREDENTIALS_JSON Google Cloud 서비스 계정 키의 JSON 문자열 (아래 참고)

  6. 변경사항 저장을 클릭합니다.

OCI 함수에 대한 Google Cloud 서비스 계정 만들기

OCI 함수가 Cloud Storage 버킷에 쓰려면 Google Cloud 서비스 계정이 필요합니다.

  1. Google Cloud 콘솔에서 IAM 및 관리자 > 서비스 계정으로 이동합니다.
  2. 서비스 계정 만들기를 클릭합니다.
  3. 다음 구성 세부정보를 제공합니다.
    • 서비스 계정 이름: oci-function-gcs-writer을 입력합니다.
    • 서비스 계정 설명: Service account for OCI Function to write Audit logs to GCS을 입력합니다.
  4. 만들고 계속하기를 클릭합니다.
  5. 이 서비스 계정에 프로젝트에 대한 액세스 권한 부여 섹션에서 다음 역할을 추가합니다.
    1. 역할 선택을 클릭합니다.
    2. 스토리지 객체 관리자를 검색하고 선택합니다.
  6. 계속을 클릭합니다.
  7. 완료를 클릭합니다.
  8. 새로 만든 서비스 계정 이메일을 클릭합니다.
  9. 키 탭으로 이동합니다.
  10. 키 추가 > 새 키 만들기를 클릭합니다.
  11. 키 유형으로 JSON을 선택합니다.
  12. 만들기를 클릭합니다.
  13. JSON 키 파일이 컴퓨터에 다운로드됩니다.
  14. JSON 키 파일을 열고 전체 콘텐츠를 복사합니다.
  15. Oracle Cloud 콘솔 함수 구성으로 돌아갑니다.

  16. JSON 콘텐츠를 GCS_CREDENTIALS_JSON 구성 변수에 붙여넣습니다.

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

서비스 계정에 Cloud Storage 버킷에 대한 쓰기 권한을 부여합니다.

  1. Cloud Storage > 버킷으로 이동합니다.
  2. 버킷 이름 (oci-audit-logs-gcs)을 클릭합니다.
  3. 권한 탭으로 이동합니다.
  4. 액세스 권한 부여를 클릭합니다.
  5. 다음 구성 세부정보를 제공합니다.
    • 주 구성원 추가: 서비스 계정 이메일 (oci-function-gcs-writer@PROJECT_ID.iam.gserviceaccount.com)을 입력합니다.
    • 역할 할당: 스토리지 객체 관리자를 선택합니다.
  6. 저장을 클릭합니다.

OCI 서비스 커넥터 허브 만들기

  1. Oracle Cloud 콘솔에 로그인합니다.
  2. 모니터링 가능성 및 관리 > 로깅 > 서비스 커넥터 허브로 이동합니다.
  3. 서비스 커넥터를 만들려는 구획을 선택합니다.
  4. 서비스 커넥터 만들기를 클릭합니다.

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

    • 서비스 커넥터 정보:

      설정 값
      커넥터 이름 audit-logs-to-gcs-connector 입력
      설명 Forward OCI Audit logs to Cloud Storage 입력
      리소스 구획 구획 선택

    • 소스 구성:

      설정 값
      소스 로깅을 선택하세요
      Compartment 감사 로그가 포함된 구획 선택
      로그 그룹 _Audit (감사 로그의 기본 로그 그룹)을 선택합니다.

  6. + 다른 로그를 클릭합니다.

  7. 구획의 감사 로그를 선택합니다 (예: _Audit_Include_Subcompartment).

    • 대상 구성:

      설정 값
      대상 함수를 선택합니다.
      함수 구획 함수가 포함된 구획 선택
      함수 적용 audit-logs-to-gcs-app 선택
      기능 oci-audit-to-gcs 선택

  8. 태스크 구성 (선택사항)으로 이동하여 기본 설정을 그대로 둡니다.

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

서비스 커넥터 허브에 대한 IAM 정책 만들기

서비스 커넥터 허브에는 함수를 호출할 수 있는 권한이 필요하며, 이 권한은 다음과 같이 만듭니다.

  1. Oracle Cloud 콘솔에서 ID 및 보안 > 정책으로 이동합니다.
  2. 서비스 커넥터 허브를 만든 구획을 선택합니다.
  3. 정책 만들기를 클릭합니다.
  4. 다음 구성 세부정보를 제공합니다.
    • 이름: service-connector-functions-policy를 입력합니다.
    • 설명: Allow Service Connector Hub to invoke Functions을 입력합니다.
    • 구획: 구획을 선택합니다.
  5. 정책 빌더 섹션에서 수동 편집기 표시를 전환합니다.

  6. 다음 정책 문을 입력합니다.

    Allow any-user to use fn-function in compartment <compartment-name> where all {request.principal.type='serviceconnector'}
Allow any-user to use fn-invocation in compartment <compartment-name> where all {request.principal.type='serviceconnector'}
    • Thinking Process <compartment-name>을 구획 이름으로 바꿉니다.

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

통합 테스트

  1. Oracle Cloud 콘솔에 로그인합니다.
  2. 감사 로그를 생성하는 작업 (예: 리소스 생성 또는 수정)을 수행합니다.
  3. 로그가 처리될 때까지 2~5분 정도 기다립니다.
  4. Google Cloud 콘솔에서 Cloud Storage > 버킷으로 이동합니다.
  5. 버킷 이름 (oci-audit-logs-gcs)을 클릭합니다.
  6. 접두사 폴더 (oci-audit-logs/)로 이동합니다.

  7. 새 .ndjson 파일이 버킷에 표시되는지 확인합니다.

Google SecOps 서비스 계정 가져오기

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

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

  1. SIEM 설정 > 피드로 이동합니다.
  2. 새 피드 추가를 클릭합니다.
  3. 단일 피드 구성을 클릭합니다.
  4. 피드 이름 필드에 피드 이름을 입력합니다(예: Oracle Cloud Audit Logs).
  5. 소스 유형으로 Google Cloud Storage V2를 선택합니다.
  6. 로그 유형으로 Oracle Cloud Infrastructure 감사 로그를 선택합니다.

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

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

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

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

Google SecOps 서비스 계정에는 Cloud Storage 버킷에 대한 두 가지 역할이 필요합니다. 로그 객체를 읽을 수 있는 스토리지 객체 뷰어와 버킷 메타데이터를 읽을 수 있는 버킷 수준 역할입니다.

  1. Cloud Storage > 버킷으로 이동합니다.
  2. 버킷 이름 (oci-audit-logs-gcs)을 클릭합니다.
  3. 권한 탭으로 이동합니다.
  4. 액세스 권한 부여를 클릭합니다.
  5. 다음 구성 세부정보를 제공합니다.
    • 주 구성원 추가: Google SecOps 서비스 계정 이메일을 붙여넣습니다.
    • 역할 할당: 다음 두 가지를 모두 선택합니다.
    • 스토리지 객체 뷰어: 로그 객체를 읽습니다.
    • 스토리지 기존 버킷 리더: 버킷 메타데이터를 읽습니다. 전송된 파일 및 빈 디렉터리 삭제 삭제 옵션을 선택한 경우 대신 스토리지 기존 버킷 작성자를 선택합니다. 이 역할은 삭제 권한도 부여합니다.

  6. 저장을 클릭합니다.

OCI 감사 로그를 수집하도록 Google SecOps에서 피드 구성

  1. SIEM 설정 > 피드로 이동합니다.
  2. 새 피드 추가를 클릭합니다.
  3. 단일 피드 구성을 클릭합니다.
  4. 피드 이름 필드에 피드 이름을 입력합니다(예: Oracle Cloud Audit Logs).
  5. 소스 유형으로 Google Cloud Storage V2를 선택합니다.
  6. 로그 유형으로 Oracle Cloud Infrastructure 감사 로그를 선택합니다.
  7. 다음을 클릭합니다.

  8. 다음 입력 파라미터의 값을 지정합니다.

    • 스토리지 버킷 URL: 프리픽스 경로가 있는 Cloud Storage 버킷 URI를 입력합니다.
    gs://oci-audit-logs-gcs/oci-audit-logs/

    • 다음과 같이 바꿉니다.

      • oci-audit-logs-gcs: Cloud Storage 버킷 이름입니다.
      • oci-audit-logs: 로그가 저장되는 선택적 프리픽스 또는 경로입니다 (루트의 경우 비워 둠).

      예:

      • 루트 버킷: gs://company-logs/
      • 프리픽스: gs://company-logs/oci-audit-logs/

      • 하위 폴더: gs://company-logs/oracle/audit/

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

      • 파일 삭제 안함: 소스에서 파일을 삭제하지 않습니다 (테스트에 권장).
      • 전송된 파일 및 빈 디렉터리 삭제: 가져오기가 완료된 후 소스에서 파일과 빈 디렉터리를 삭제합니다.

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

    • 애셋 네임스페이스: 애셋 네임스페이스입니다.

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

  9. 다음을 클릭합니다.

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

방법 2: 웹훅

이 방법을 사용하면 커넥터 허브가 감사 로그 항목 배치를 OCI 함수에 전달하고 함수는 HTTPS를 통해 Google SecOps 웹훅 피드로 전송합니다. Cloud Storage 버킷은 관련이 없습니다.

Google SecOps에서 웹훅 피드 만들기

피드 만들기

  1. SIEM 설정 > 피드로 이동합니다.
  2. 새 피드 추가를 클릭합니다.
  3. 다음 페이지에서 단일 피드 구성을 클릭합니다.
  4. 피드 이름 필드에 피드 이름을 입력합니다(예: OCI Audit Logs Webhook).
  5. 소스 유형으로 웹훅을 선택합니다.
  6. 로그 유형으로 Oracle Cloud Infrastructure 감사 로그를 선택합니다.
  7. 다음을 클릭합니다.
  8. 다음 입력 파라미터의 값을 지정합니다.
    • 구분자 분할: \n를 입력합니다. 함수는 각 로그 항목을 자체 줄로 전송합니다.
    • 애셋 네임스페이스: 애셋 네임스페이스입니다.
    • 수집 라벨: 이 피드의 이벤트에 적용할 라벨입니다.
  9. 다음을 클릭합니다.
  10. 확정 화면에서 새 피드 구성을 검토한 다음 제출을 클릭합니다.

보안 비밀 키 생성 및 저장

  1. 피드 세부정보 페이지에서 보안 비밀 키 생성을 클릭합니다.

  2. 보안 비밀 키를 복사하여 안전하게 저장합니다.

피드 엔드포인트 URL 가져오기

  1. 피드의 세부정보 탭으로 이동합니다.

  2. 엔드포인트 정보 섹션에서 피드 엔드포인트 URL을 복사합니다. URL의 형식은 다음과 같습니다.

    https://<REGION>-chronicle.googleapis.com/v1alpha/projects/<PROJECT_NUMBER>/locations/<REGION>/instances/<INSTANCE_ID>/feeds/<FEED_ID>:importPushLogs

  3. 완료를 클릭합니다.

Google Cloud API 키 만들기

  1. Google Cloud 콘솔 사용자 인증 정보 페이지로 이동합니다.
  2. Google SecOps 인스턴스에 연결된 프로젝트를 선택합니다.
  3. 사용자 인증 정보 만들기 > API 키를 클릭합니다.
  4. API 키 수정을 클릭하여 키를 제한합니다.
  5. 이름 필드에 설명이 포함된 이름을 입력합니다 (예: oci-audit-webhook).
  6. API 제한사항에서 키 제한을 선택한 후 API 선택 목록에서 Chronicle API를 선택합니다.
  7. 저장을 클릭합니다.
  8. API 키 값을 복사하여 안전하게 저장합니다.

웹훅에 로그를 전달하는 OCI 함수 만들기

애플리케이션 만들기

  1. Oracle Cloud 콘솔에 로그인합니다.
  2. 탐색 메뉴를 열고 개발자 서비스 > 함수 > 애플리케이션을 선택합니다.
  3. 함수 애플리케이션을 만들 구획을 선택합니다.
  4. 애플리케이션 만들기를 클릭합니다.
  5. 다음 구성 세부정보를 제공합니다.
    • 이름: audit-logs-to-secops-app를 입력합니다.
    • VCN: 가상 클라우드 네트워크를 선택합니다.
    • 서브넷: 인터넷 액세스가 있는 서브넷(인터넷 게이트웨이가 있는 공개 서브넷 또는 NAT 게이트웨이가 있는 비공개 서브넷)을 선택합니다. 함수는 HTTPS (포트 443)를 통해 Google SecOps 엔드포인트에 도달해야 합니다.
  6. 만들기를 클릭합니다.

함수 만들기 및 배포

  1. 애플리케이션이 생성되면 시작하기를 클릭하고 안내에 따라 Fn CLI를 사용하여 OCI Cloud Shell 또는 로컬 개발 환경을 설정합니다.

  2. Python 함수를 초기화합니다.

    fn init --runtime python oci-audit-to-secops
cd oci-audit-to-secops

  3. func.py의 내용을 다음 코드로 바꿉니다.

    import io
import json
import logging
import os
import random
import time
from email.utils import parsedate_to_datetime

import requests
from fdk import response

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger()

# The webhook accepts at most 4 MB per request and 1 MB per log line.
# Connector Hub delivers batches of up to 6 MB, so a batch is sent as several
# requests, each made of whole log entries.
MAX_REQUEST_BYTES = 3 * 1024 * 1024
MAX_LINE_BYTES = 1024 * 1024

MAX_ATTEMPTS = 5
RETRYABLE_STATUS = {429, 500, 502, 503, 504}
# OCI stops the function at its configured timeout (300 seconds). Stop retrying
# early enough to fail cleanly so that Connector Hub retries the whole batch.
DEADLINE_SECONDS = 240

def to_lines(body):
    """Serialize each log entry from the Connector Hub batch as one JSON line."""
    entries = body if isinstance(body, list) else [body]
    lines = []
    for entry in entries:
        line = json.dumps(entry, ensure_ascii=False, separators=(',', ':')).encode('utf-8')
        if len(line) > MAX_LINE_BYTES:
            entry_id = entry.get('id') if isinstance(entry, dict) else None
            logger.error(f'Skipping log entry {entry_id}: {len(line)} bytes exceeds the 1 MB line limit')
            continue
        lines.append(line)
    return lines

def to_requests(lines):
    """Group whole lines into newline-delimited request bodies below MAX_REQUEST_BYTES."""
    chunk, size = [], 0
    for line in lines:
        if chunk and size + len(line) + 1 > MAX_REQUEST_BYTES:
            yield b'\n'.join(chunk), len(chunk)
            chunk, size = [], 0
        chunk.append(line)
        size += len(line) + 1
    if chunk:
        yield b'\n'.join(chunk), len(chunk)

def retry_after_seconds(value):
    """Parse a Retry-After header given either in seconds or as an HTTP date."""
    if not value:
        return None
    try:
        return max(0.0, float(value))
    except ValueError:
        pass
    try:
        return max(0.0, parsedate_to_datetime(value).timestamp() - time.time())
    except (TypeError, ValueError, IndexError, OverflowError):
        return None

def post(session, url, headers, payload, deadline):
    """Send one request, retrying throttling and server errors until the deadline."""
    status, detail = None, ''
    for attempt in range(1, MAX_ATTEMPTS + 1):
        remaining = deadline - time.monotonic()
        if remaining < 5:
            break
        wait = None
        try:
            resp = session.post(url, data=payload, headers=headers, timeout=(10, min(60, remaining)))
        except requests.RequestException as e:
            status, detail = None, str(e)
        else:
            if 200 <= resp.status_code < 300:
                return
            status, detail = resp.status_code, resp.text[:500]
            if status not in RETRYABLE_STATUS:
                raise RuntimeError(f'Webhook rejected the request with HTTP {status}: {detail}')
            wait = retry_after_seconds(resp.headers.get('Retry-After'))
        if wait is None:
            wait = min(30, 2 ** attempt) + random.uniform(0, 1)
        if attempt == MAX_ATTEMPTS or time.monotonic() + wait >= deadline:
            break
        logger.warning(f'Attempt {attempt} failed (HTTP {status}): {detail}. Retrying in {wait:.1f} seconds')
        time.sleep(wait)
    raise RuntimeError(f'Webhook request failed (HTTP {status}): {detail}')

def handler(ctx, data: io.BytesIO = None):
    """Forward a Connector Hub batch of OCI Audit log entries to the webhook feed.

    Any exception fails the invocation, and Connector Hub then retries the batch.
    """
    deadline = time.monotonic() + DEADLINE_SECONDS

    url = os.environ.get('SECOPS_WEBHOOK_URL')
    api_key = os.environ.get('SECOPS_API_KEY')
    secret_key = os.environ.get('SECOPS_SECRET_KEY')
    if not all([url, api_key, secret_key]):
        raise RuntimeError('Missing configuration: SECOPS_WEBHOOK_URL, SECOPS_API_KEY and SECOPS_SECRET_KEY are required')

    raw = data.getvalue() if data else b''
    if not raw.strip():
        logger.info('Received an empty batch')
        return response.Response(
            ctx, response_data=json.dumps({'records': 0}),
            headers={'Content-Type': 'application/json'}
        )

    lines = to_lines(json.loads(raw))
    headers = {
        'Content-Type': 'application/json',
        'X-goog-api-key': api_key,
        'X-Webhook-Access-Key': secret_key,
    }

    sent = 0
    with requests.Session() as session:
        for payload, count in to_requests(lines):
            post(session, url, headers, payload, deadline)
            sent += count

    logger.info(f'Forwarded {sent} log entries')
    return response.Response(
        ctx, response_data=json.dumps({'records': sent}),
        headers={'Content-Type': 'application/json'}
    )

  4. requirements.txt의 내용을 다음 종속 항목으로 바꿉니다.

    fdk>=0.1.0
requests>=2.32.0

  5. func.yaml을 열고 다음 줄을 추가하여 함수 제한 시간을 기본값인 30초에서 최대 300초로 늘립니다.

    timeout: 300

  6. OCI에 함수를 배포합니다.

    fn -v deploy --app audit-logs-to-secops-app

함수 환경 변수 구성

  1. Oracle Cloud 콘솔에서 개발자 서비스 > 함수 > 애플리케이션으로 이동합니다.
  2. 애플리케이션 (audit-logs-to-secops-app)을 클릭합니다.
  3. 함수 이름 (oci-audit-to-secops)을 클릭합니다.
  4. 구성을 클릭합니다.

  5. 다음 구성 변수를 추가합니다.

    키 값
    SECOPS_WEBHOOK_URL 피드 엔드포인트 URL
    SECOPS_API_KEY Google Cloud API 키
    SECOPS_SECRET_KEY 피드 보안 비밀 키

  6. 변경사항 저장을 클릭합니다.

함수 로그 사용 설정

  1. 애플리케이션 목록 페이지에서 audit-logs-to-secops-app를 선택합니다.
  2. 모니터링 탭을 선택하고 로그 섹션으로 이동합니다.
  3. 함수 호출 로그 카테고리의 작업 메뉴 (점 3개)에서 로그 사용 설정을 선택합니다.
  4. 기본 구획, 로그 그룹, 로그 이름을 유지하거나 필요에 따라 변경합니다.
  5. 로그 사용 설정을 선택합니다.

OCI Connector Hub 커넥터 만들기

  1. 탐색 메뉴를 열고 분석 및 AI를 선택합니다. 메시지에서 커넥터 허브를 선택합니다.
  2. 커넥터를 만들려는 구획을 선택합니다.
  3. 커넥터 만들기를 클릭합니다.

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

    • 커넥터 정보:

      설정 값
      커넥터 이름 audit-logs-to-secops-connector 입력
      설명 Forward OCI Audit logs to a webhook feed 입력
      리소스 구획 구획 선택

    • 소스 구성:

      설정 값
      소스 로깅을 선택하세요
      Compartment 감사 로그가 포함된 구획 선택
      로그 그룹 _Audit (감사 로그의 기본 로그 그룹)을 선택합니다.

  5. + 다른 로그를 클릭합니다.

  6. 구획의 감사 로그를 선택합니다 (예: _Audit_Include_Subcompartment).

    • 대상 구성:

      설정 값
      대상 함수를 선택합니다.
      함수 구획 함수가 포함된 구획 선택
      함수 적용 audit-logs-to-secops-app 선택
      기능 oci-audit-to-secops 선택

  7. 태스크 구성 (선택사항)으로 이동하여 기본 설정을 그대로 둡니다.

  8. 정책을 만들라는 메시지가 표시되면 (서비스 커넥터를 만들거나 업데이트하는 액세스 권한에 필요) 만들기를 선택합니다.

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

커넥터 허브의 IAM 정책 만들기

커넥터 허브에는 함수를 호출할 수 있는 권한이 필요합니다.

  1. Oracle Cloud 콘솔에서 ID 및 보안 > 정책으로 이동합니다.
  2. 커넥터를 만든 구획을 선택합니다.
  3. 정책 만들기를 클릭합니다.
  4. 다음 구성 세부정보를 제공합니다.
    • 이름: connector-hub-secops-function-policy를 입력합니다.
    • 설명: Allow Connector Hub to invoke the webhook forwarding function을 입력합니다.
    • 구획: 구획을 선택합니다.
  5. 정책 빌더 섹션에서 수동 편집기 표시를 전환합니다.

  6. 다음 정책 문을 입력합니다.

    Allow any-user to use fn-function in compartment <compartment-name> where all {request.principal.type='serviceconnector'}
Allow any-user to use fn-invocation in compartment <compartment-name> where all {request.principal.type='serviceconnector'}

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

함수가 로그를 전송하는 방법

  • 배치 크기: 커넥터 허브는 최대 6MB의 함수 배치를 전송하고 웹훅은 요청당 최대 4MB를 허용합니다. 함수는 각 배치를 최대 3MB의 요청으로 분할합니다. 로그 항목 사이에서만 분할되고 항목 내에서는 분할되지 않으며 각 항목은 하나의 이벤트로 Google SecOps에 도착합니다.
  • 재시도: 함수는 지수 백오프를 사용하여 HTTP 429 및 5xx 응답을 재시도하고 Retry-After 헤더를 적용합니다. 다른 오류가 발생하거나 5번 시도한 후에도 요청이 실패하면 호출이 실패합니다. 그러면 커넥터 허브는 전체 배치를 다시 시도하고 성공할 때까지 이후 배치를 이동하지 않습니다.
  • 보관: 커넥터 허브의 Logging 소스는 전송되지 않은 데이터를 24시간 동안 보관합니다. 전송이 계속 실패하면 전송되지 않은 가장 오래된 로그가 손실되므로 함수 로그에 보고된 오류를 즉시 수정하세요.
  • 중복: 커넥터 허브는 각 배치를 최소 한 번 전송합니다. 일부가 이미 전송된 후 배치가 재시도되면 Google SecOps에서 해당 항목을 두 번 수신할 수 있습니다.
  • 크기가 큰 항목: 1MB보다 큰 로그 항목은 웹훅 줄 한도를 초과합니다. 함수는 이를 건너뛰고 ID를 함수 로그에 기록합니다.

통합 확인

통합을 확인하려면 다음을 수행합니다.

  1. Oracle Cloud 콘솔에서 감사 로그를 생성하는 작업 (예: 리소스 만들기 또는 수정)을 수행합니다.
  2. 커넥터 허브에서 로그를 전송할 때까지 2~5분 정도 기다립니다.
  3. 사용 설정한 함수 호출 로그를 열고 Forwarded N log entries 메시지가 포함되어 있는지 확인합니다. 오류 메시지에는 Google SecOps에서 반환한 HTTP 상태가 포함됩니다.

  4. Google SecOps에서 OCI_AUDIT 로그 유형에 대한 원시 로그 검색을 실행하고 감사 이벤트가 도착하는지 확인합니다.

UDM 매핑 테이블

로그 필드 UDM 매핑 로직
accept_encoding_field additional.fields 병합됨
accept_field additional.fields 병합됨
additional_app additional.fields 병합됨
additional_content_type additional.fields 병합됨
additional_sort_by additional.fields 병합됨
additionaldetails_data additional.fields 병합됨
allow_methods_field additional.fields 병합됨
auth_type_label additional.fields 병합됨
authorization_field additional.fields 병합됨
backend_connect_time_field additional.fields 병합됨
backend_processing_time_field additional.fields 병합됨
caller_id_label additional.fields 병합됨
caller_name_label additional.fields 병합됨
cloud_event_versiom_field additional.fields 병합됨
compartment_id_field additional.fields 병합됨
compartment_name_field additional.fields 병합됨
connection_field additional.fields 병합됨
content_length_field additional.fields 병합됨
content_type_field additional.fields 병합됨
event_grouping_id_field additional.fields 병합됨
event_name_label additional.fields 병합됨
event_resource_label additional.fields 병합됨
event_source_label additional.fields 병합됨
event_type_label additional.fields 병합됨
event_type_versiom_field additional.fields 병합됨
has_compariment_id additional.fields 매핑됨: true → additional_app
has_sort_by additional.fields 매핑됨: true → additional_sort_by
listener_name_field additional.fields 병합됨
log_group_id_field additional.fields 병합됨
log_id_field additional.fields 병합됨
message additional.fields 매핑된 값 (총 41개, 예: \/ → tenant_id_label, \/ → auth_type_label, \/ → `ca…
opc_request_id_field additional.fields 병합됨
origin_field additional.fields 병합됨
param0_field additional.fields 병합됨
param1_field additional.fields 병합됨
referer_field additional.fields 병합됨
request_processing_time_field additional.fields 병합됨
resource_id_field additional.fields 병합됨
routing_rules_engine_errors_field additional.fields 병합됨
routing_rules_matched_rule_field additional.fields 병합됨
routing_rules_rule_hits_field additional.fields 병합됨
routing_rules_rule_misses_field additional.fields 병합됨
source_field additional.fields 병합됨
tenant_id_label additional.fields 병합됨
column3 extensions.auth.type 매핑됨: login successful → AUTHTYPE_UNSPECIFIED
message extensions.auth.type 매핑됨: \/ → AUTHTYPE_UNSPECIFIED
column3 metadata.description 직접 매핑됨
data.message metadata.description 직접 매핑됨
column1 metadata.event_timestamp yyyy-MM-dd HH:mm:ss(으)로 파싱됨
data_event_time metadata.event_timestamp yyyy-MM-ddTHH:mm:ss.SSSSSSZ(으)로 파싱됨
time metadata.event_timestamp ISO8601(으)로 파싱됨
metadata_event_type metadata.event_type 직접 매핑됨
type metadata.product_event_type 직접 매핑됨
data_event_id metadata.product_log_id 직접 매핑됨
id metadata.product_log_id 직접 매핑됨
message metadata.product_name 매핑됨: \/ → OCI_AUDIT
specversion metadata.product_version 직접 매핑됨
message metadata.vendor_name 매핑됨: \/ → Oracle
message network.application_protocol 매핑됨: \/ → HTTPS, \/ → HTTP
protocol network.application_protocol 매핑됨: (?i)HTTPS → HTTPS, (?i)HTTP → HTTP
version network.application_protocol_version 직접 매핑됨
data_request_action network.http.method 직접 매핑됨
method network.http.method 직접 매핑됨
data_identity_userAgent network.http.parsed_user_agent 이름 변경/매핑됨
data_request_agent network.http.parsed_user_agent 이름 변경/매핑됨
url network.http.referral_url 직접 매핑됨
data.backendStatusCode network.http.response_code 직접 매핑됨
data_response_status network.http.response_code 이름 변경/매핑됨
data.userAgent network.http.user_agent 직접 매핑됨
data_identity_userAgent network.http.user_agent 직접 매핑됨
data_request_agent network.http.user_agent 직접 매핑됨
data.receivedBytes network.received_bytes 직접 매핑됨
message network.received_bytes 매핑됨: \/ → uinteger
data.sentBytes network.sent_bytes 직접 매핑됨
message network.sent_bytes 매핑됨: \/ → uinteger
data_identity_consoleSessionId network.session_id 이름 변경/매핑됨
data.sslCipher network.tls.cipher 직접 매핑됨
data.sslProtocol network.tls.version 직접 매핑됨
data.host principal.asset.hostname 직접 매핑됨
hostname principal.asset.hostname 직접 매핑됨
column6 principal.asset.ip 병합됨
data.forwardedForAddr principal.asset.ip 병합됨
data_identity_ipAddress principal.asset.ip 매핑됨: ^(?:[0-9]{1,3}[.]){3}[0-9]{1,3}$ → data_identity_ipAddress
data_request_origin principal.asset.ip 병합됨
ip1 principal.asset.ip 병합됨
ip2 principal.asset.ip 병합됨
message principal.asset.ip 매핑된 값 (총 7개, 예: \/ → column6, \/ → data_request_origin, \/ → ip2)
src_ip principal.asset.ip 병합됨
data.host principal.hostname 직접 매핑됨
hostname principal.hostname 직접 매핑됨
column6 principal.ip 병합됨
data.forwardedForAddr principal.ip 병합됨
data_identity_ipAddress principal.ip 매핑됨: ^(?:[0-9]{1,3}[.]){3}[0-9]{1,3}$ → data_identity_ipAddress
data_request_origin principal.ip 병합됨
ip1 principal.ip 병합됨
ip2 principal.ip 병합됨
message principal.ip 매핑된 값 (총 7개, 예: \/ → column6, \/ → data_request_origin, \/ → ip2)
src_ip principal.ip 병합됨
data_request_headers_sec-ch-ua-platform_0 principal.platform 매핑됨: (?i)Linux → LINUX, (?i)windows → WINDOWS, (?i)mac/ios → MAC
message principal.platform 매핑됨: \/ → LINUX, \/ → WINDOWS, \/ → MAC
originalConnection.sourcePort principal.port 직접 매핑됨
src_port principal.port 직접 매핑됨
data_request_headers_oci-original-url_0 principal.url 직접 매핑됨
credentials_label principal.user.attribute.labels 병합됨
message principal.user.attribute.labels 매핑됨: \/ → credentials_label
data.identity.principalName principal.user.user_display_name 직접 매핑됨
data.username principal.user.user_display_name 직접 매핑됨
data.identity.principalId principal.user.userid 직접 매핑됨
data.principalId principal.user.userid 직접 매핑됨
column3 security_result 매핑됨: login successful → security_result
message security_result 매핑됨: \/ → security_result
allow_action security_result.action 병합됨
column3 security_result.action 매핑됨: login successful → allow_action
message security_result.action 매핑됨: \/ → allow_action
credential_type_field security_result.detection_fields 병합됨
message security_result.detection_fields 매핑됨: \/ → credential_type_field
data.response.message security_result.summary 직접 매핑됨
message target.asset.ip 매핑됨: \/ → originalConnection.destinationIp, \/ → tar_ip
originalConnection.destinationIp target.asset.ip 병합됨
tar_ip target.asset.ip 병합됨
message target.ip 매핑됨: \/ → originalConnection.destinationIp, \/ → tar_ip
originalConnection.destinationIp target.ip 병합됨
tar_ip target.ip 병합됨
originalConnection.destinationPort target.port 직접 매핑됨
tar_port target.port 직접 매핑됨
message target.resource.attribute.labels 매핑됨: \/ → namespace_label
namespace_label target.resource.attribute.labels 병합됨
type target.resource.attribute.labels 매핑됨: listretentionrules → namespace_label
data.resourceId target.resource.name 직접 매핑됨
data.resourceName target.resource.name 직접 매핑됨
bucketId target.resource.product_object_id 직접 매핑됨
data.resourceId target.resource.product_object_id 직접 매핑됨
data_request_id target.resource.product_object_id 직접 매핑됨
message target.resource.resource_type 매핑됨: \/ → STORAGE_BUCKET
type target.resource.resource_type 매핑됨: listretentionrules → STORAGE_BUCKET
data_request_path target.url 직접 매핑됨
column2 target.user.email_addresses 병합됨
message target.user.email_addresses 매핑됨: \/ → column2
해당 사항 없음 extensions.auth.type 상수: AUTHTYPE_UNSPECIFIED
해당 사항 없음 metadata.product_name 상수: OCI_AUDIT
해당 사항 없음 metadata.vendor_name 상수: Oracle
해당 사항 없음 network.application_protocol 상수: HTTPS
해당 사항 없음 principal.platform 상수: LINUX
해당 사항 없음 target.resource.resource_type 상수: STORAGE_BUCKET

변경 로그

이 파서의 변경 로그 보기

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