# Журнал доступа ingress-шлюза Istio: envoyOtelAls (провайдер audit-access в meshConfig istiod) → OTLP gRPC → Kafka. apiVersion: helm.toolkit.fluxcd.io/v2 kind: HelmRelease metadata: name: access-collector namespace: security-audit spec: interval: 10m timeout: 10m chart: spec: chart: opentelemetry-collector-prod version: "0.117.1" sourceRef: kind: HelmRepository name: yc-oci-charts namespace: flux-system interval: 10m install: remediation: retries: 3 upgrade: remediation: retries: 3 values: mode: deployment replicaCount: 2 fullnameOverride: access-collector # Тот же образ, что у audit-collector: otlp receiver, transform, kafka exporter есть в сборке. image: repository: cr.yandex/crp3ccidau046kdj8g9q/opentelemetry-collector/opentelemetry-collector-contrib tag: "0.121.0-amd64" command: name: otelcol-contrib imagePullSecrets: - name: regcred podDisruptionBudget: enabled: true minAvailable: 1 # Мягкая: при одном доступном узле вторая реплика всё равно встанет. affinity: podAntiAffinity: preferredDuringSchedulingIgnoredDuringExecution: - weight: 100 podAffinityTerm: topologyKey: kubernetes.io/hostname labelSelector: matchLabels: security-audit/collector: access clusterRole: create: false serviceAccount: create: true name: access-collector # bound SA роли Vault security-audit (вместе с audit-collector) presets: logsCollection: {enabled: false} hostMetrics: {enabled: false} kubernetesAttributes: {enabled: false} kubeletMetrics: {enabled: false} kubernetesEvents: {enabled: false} clusterMetrics: {enabled: false} ports: otlp: enabled: true containerPort: 4317 servicePort: 4317 protocol: TCP appProtocol: grpc # Istio выбирает протокол по appProtocol # События Zitadel Actions v2 (Target restAsync) — обычный JSON по HTTP, приёмник webhookevent. zitadel-events: enabled: true containerPort: 8088 servicePort: 8088 protocol: TCP appProtocol: http # Журнал аудита Vault от sidecar'а audit-shipper в поде Vault (OTLP gRPC). otlp-vault: enabled: true containerPort: 4319 servicePort: 4319 protocol: TCP appProtocol: grpc otlp-http: {enabled: false} jaeger-compact: {enabled: false} jaeger-thrift: {enabled: false} jaeger-grpc: {enabled: false} zipkin: {enabled: false} metrics: {enabled: false} service: enabled: true podSecurityContext: runAsNonRoot: true runAsUser: 10001 runAsGroup: 10001 seccompProfile: {type: RuntimeDefault} securityContext: runAsNonRoot: true runAsUser: 10001 readOnlyRootFilesystem: true allowPrivilegeEscalation: false privileged: false capabilities: drop: [ALL] resources: requests: {cpu: 100m, memory: 128Mi} limits: {cpu: 1000m, memory: 512Mi} podLabels: security-audit/collector: access # селектор NetworkPolicy и anti-affinity sidecar.istio.io/inject: "true" # mTLS внутри mesh (п.30); namespace без istio-injection podAnnotations: sidecar.istio.io/inject: "true" # Kafka exporter подключается к брокерам на старте: без готового istio-proxy коллектор упадёт. proxy.istio.io/config: '{"holdApplicationUntilProxyStarts": true}' # Паттерн контура для Vault + sidecar (apps/*/base): трафик агента к Vault мимо istio-proxy. traffic.sidecar.istio.io/excludeOutboundPorts: "8200" vault.hashicorp.com/agent-inject: "true" vault.hashicorp.com/agent-init-first: "true" vault.hashicorp.com/agent-pre-populate-only: "true" vault.hashicorp.com/agent-requests-cpu: "25m" vault.hashicorp.com/agent-limits-cpu: "100m" vault.hashicorp.com/auth-path: auth/kubernetes vault.hashicorp.com/role: security-audit # Многострочные шаблоны — см. helmrelease-collector.yaml (toYaml | tpl в чарте). vault.hashicorp.com/agent-inject-secret-kafka-username: secrets/data/security-audit/kafka vault.hashicorp.com/agent-inject-template-kafka-username: |- {{ "{{- with secret \"secrets/data/security-audit/kafka\" -}}" }} {{ "{{ .Data.data.username }}" }} {{ "{{- end -}}" }} vault.hashicorp.com/agent-inject-secret-kafka-password: secrets/data/security-audit/kafka vault.hashicorp.com/agent-inject-template-kafka-password: |- {{ "{{- with secret \"secrets/data/security-audit/kafka\" -}}" }} {{ "{{ .Data.data.password }}" }} {{ "{{- end -}}" }} vault.hashicorp.com/agent-inject-secret-kafka-ca.crt: secrets/data/security-audit/kafka vault.hashicorp.com/agent-inject-template-kafka-ca.crt: |- {{ "{{- with secret \"secrets/data/security-audit/kafka\" -}}" }} {{ "{{ .Data.data.ca_crt }}" }} {{ "{{- end -}}" }} # Контур-специфичное (CONTOUR, KAFKA_BROKERS) задаётся в оверлее через extraEnvs. alternateConfig: extensions: health_check: {endpoint: "${env:MY_POD_IP}:13133"} receivers: otlp: protocols: grpc: endpoint: "${env:MY_POD_IP}:4317" max_recv_msg_size_mib: 16 # Подпись HMAC (X-ZITADEL-Signature) не проверяется: доступ ограничен mTLS + AuthorizationPolicy # (только SA zitadel) + NetworkPolicy. Отвечает 200 после ConsumeLogs, т. е. после приёма в batch. # Отдельный порт: доступ на 4319 только для SA Vault (AuthorizationPolicy), 4317 — только шлюз. otlp/vault: protocols: grpc: endpoint: "${env:MY_POD_IP}:4319" webhookevent/zitadel: endpoint: "${env:MY_POD_IP}:8088" path: /zitadel/events health_path: /zitadel/health processors: memory_limiter: {check_interval: 1s, limit_mib: 400, spike_limit_mib: 100} transform/access: error_mode: ignore log_statements: - context: log statements: # labels из envoyOtelAls приходят атрибутами — делаем из них тело (в Kafka уходит body как JSON) - set(body, attributes) - set(body["sarex_contour"], "${env:CONTOUR}") # Envoy пишет "-" для пустых значений - set(body["user_sub"], "") where body["user_sub"] == "-" - set(body["user_org"], "") where body["user_org"] == "-" - set(body["trace_id"], "") where body["trace_id"] == "-" # маскирование: токены в самом пути (публичные ссылки и т. п.) - replace_pattern(body["path"], "(?i)(token|secret|key|signature)/[^/]+", "$$1/***") - replace_pattern(body["path"], "[A-Za-z0-9_-]{20,}\\.[A-Za-z0-9_-]{20,}\\.[A-Za-z0-9_-]{10,}", "***jwt***") # классификация - set(body["sarex_action"], "read") where body["method"] == "GET" or body["method"] == "HEAD" or body["method"] == "OPTIONS" - set(body["sarex_action"], "mutate") where body["method"] == "POST" or body["method"] == "PUT" or body["method"] == "PATCH" or body["method"] == "DELETE" - set(body["sarex_outcome"], "applied") where Int(body["response_code"]) >= 200 and Int(body["response_code"]) < 400 - set(body["sarex_outcome"], "denied_authn") where body["response_code"] == "401" - set(body["sarex_outcome"], "denied_authz") where body["response_code"] == "403" - set(body["sarex_outcome"], "invalid") where Int(body["response_code"]) >= 400 and Int(body["response_code"]) < 500 and body["response_code"] != "401" and body["response_code"] != "403" - set(body["sarex_outcome"], "error") where Int(body["response_code"]) >= 500 or body["response_code"] == "0" - set(body["sarex_actor_type"], "anonymous") - set(body["sarex_actor_type"], "human") where body["user_sub"] != "" transform/idp: error_mode: ignore log_statements: - context: log statements: - set(body, ParseJSON(body)) # Не JSON — тело целиком заменяется маркером: сырой текст мог бы унести event_payload в Kafka (п.23). - 'set(body, {"sarex_kind": "idp_event", "sarex_error": "unparsed"}) where not IsMap(body)' # В 4.12.2 event_payload — json.RawMessage (объект); base64-строку тоже поддерживаем - set(body["event_payload"], ParseJSON(Decode(body["event_payload"], "base64"))) where IsString(body["event_payload"]) # allowlist (п.23): нужные ИБ поля копируются явно, payload удаляется ЦЕЛИКОМ # (в событиях бывают хеши паролей, OTP-секреты, client secret, ПДн) # TODO(verify): список полей по типам событий уточнить по реальным событиям на стенде - set(body["details"]["user_name"], body["event_payload"]["userName"]) where IsMatch(body["event_type"], "^user\\.human\\.(added|selfregistered)$") - set(body["details"]["role_keys"], body["event_payload"]["roleKeys"]) where IsMatch(body["event_type"], "^user\\.grant\\.") - set(body["details"]["project_id"], body["event_payload"]["projectId"]) where IsMatch(body["event_type"], "^user\\.grant\\.") - delete_key(body, "event_payload") # обогащение и классификация - set(body["sarex_kind"], "idp_event") - set(body["sarex_contour"], "${env:CONTOUR}") - set(body["sarex_outcome"], "applied") - set(body["sarex_outcome"], "denied_authn") where IsMatch(body["event_type"], "check\\.failed$") - set(body["sarex_actor_type"], "human") - set(body["sarex_actor_type"], "system") where body["userID"] == "" or body["userID"] == nil batch: {send_batch_size: 1000, send_batch_max_size: 2000, timeout: 2s} exporters: # Общие поля Kafka — YAML-якоря (&kafka_*), kafka/idp ссылается на них (*kafka_*). kafka/access: brokers: &kafka_brokers ["${env:KAFKA_BROKERS}"] protocol_version: &kafka_protocol_version 2.8.0 topic: security.mesh-access encoding: raw # в сообщение — только body (JSON записи), без OTLP-обёртки auth: &kafka_auth sasl: {mechanism: SCRAM-SHA-512, username: "${file:/vault/secrets/kafka-username}", password: "${file:/vault/secrets/kafka-password}"} tls: {insecure: false, ca_file: /vault/secrets/kafka-ca.crt} producer: &kafka_producer {required_acks: -1, compression: zstd, max_message_bytes: 1048576} # Очередь в памяти (диска у Deployment нет): при падении пода неотправленные записи теряются. sending_queue: &kafka_sending_queue {enabled: true, queue_size: 50000, num_consumers: 4} retry_on_failure: &kafka_retry {enabled: true, initial_interval: 5s, max_interval: 60s, max_elapsed_time: 0} kafka/idp: brokers: *kafka_brokers protocol_version: *kafka_protocol_version topic: security.idp-events encoding: raw auth: *kafka_auth producer: *kafka_producer sending_queue: *kafka_sending_queue # своя очередь у каждого экспортёра, общие только настройки retry_on_failure: *kafka_retry # Body уже JSON-объект (json_parser в audit-shipper, sarex_* там же) — transform не нужен. kafka/vault: brokers: *kafka_brokers protocol_version: *kafka_protocol_version topic: security.vault-audit encoding: raw auth: *kafka_auth producer: *kafka_producer sending_queue: *kafka_sending_queue retry_on_failure: *kafka_retry service: extensions: [health_check] pipelines: logs/access: receivers: [otlp] processors: [memory_limiter, transform/access, batch] exporters: [kafka/access] logs/idp: receivers: [webhookevent/zitadel] processors: [memory_limiter, transform/idp, batch] exporters: [kafka/idp] logs/vault: receivers: [otlp/vault] processors: [memory_limiter, batch] exporters: [kafka/vault] telemetry: logs: {level: info} metrics: {address: "${env:MY_POD_IP}:8888"}