Coletar registros de auditoria do Slack

Compatível com:

Este documento explica como ingerir registros de auditoria do Slack no Google Security Operations usando o Amazon S3. Primeiro, o analisador normaliza os valores booleanos e limpa os campos predefinidos. Em seguida, ele analisa o campo "message" como JSON, descartando mensagens que não são JSON. Dependendo da presença de campos específicos (date_create e user_id), o analisador aplica uma lógica diferente para mapear campos de registros brutos no UDM, incluindo metadados, principal, rede, destino e informações sobre, e cria um resultado de segurança.

Antes de começar

Verifique se você tem os pré-requisitos a seguir:

  • Instância do Google SecOps
  • Acesso privilegiado ao locatário do Slack Enterprise Grid e ao Admin Console
  • Acesso privilegiado à AWS (S3, IAM, Lambda, EventBridge)

Coletar os pré-requisitos do Slack (ID do app, token OAuth, ID da organização)

  1. Faça login no Admin Console do Slack.
  2. Acesse https://api.slack.com/apps e clique em Criar novo app > Do zero.
  3. Insira um Nome do app exclusivo e selecione seu espaço de trabalho do Slack.
  4. Clique em Criar app.
  5. Acesse OAuth e permissões na barra lateral esquerda.
  6. Acesse a seção Escopos e adicione o seguinte escopo de token de usuário: auditlogs:read
  7. Clique em Instalar no Workspace > Permitir.
  8. Depois de instalado, acesse Apps no nível da organização.
  9. Clique em Instalar na organização.
  10. Autorize o app com uma conta de proprietário/administrador da organização.
  11. Copie e salve com segurança o token OAuth do usuário que começa com xoxp- (esse é o SLACK_AUDIT_TOKEN).
  12. Anote o ID da organização, que pode ser encontrado no Admin Console do Slack em Configurações e permissões > Configurações da organização.

Configurar o bucket do AWS S3 e o IAM para o Google SecOps

  1. Crie um bucket do Amazon S3 seguindo este guia do usuário: Como criar um bucket
  2. Salve o Nome e a Região do bucket para referência futura (por exemplo, slack-audit-logs).
  3. Crie um usuário seguindo este guia: Como criar um usuário do IAM.
  4. Selecione o usuário criado.
  5. Selecione a guia Credenciais de segurança.
  6. Clique em Criar chave de acesso na seção Chaves de acesso.
  7. Selecione Serviço de terceiros como o Caso de uso.
  8. Clique em Próxima.
  9. Opcional: adicione uma tag de descrição.
  10. Clique em Criar chave de acesso.
  11. Clique em Fazer o download do arquivo CSV para salvar a chave de acesso e a chave de acesso secreta para uso posterior.
  12. Clique em Concluído.
  13. Selecione a guia Permissões.
  14. Clique em Adicionar permissões na seção Políticas de permissões.
  15. Selecione Adicionar permissões.
  16. Selecione Anexar políticas diretamente.
  17. Pesquise e selecione a política AmazonS3FullAccess.
  18. Clique em Próxima.
  19. Clique em Adicionar permissões

Configurar a política e o papel do IAM para uploads do S3

  1. No console da AWS, acesse IAM > Políticas > Criar política > guia JSON.

  2. Insira a seguinte política:

    {
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "AllowPutObjects",
      "Effect": "Allow",
      "Action": "s3:PutObject",
      "Resource": "arn:aws:s3:::slack-audit-logs/*"
    },
    {
      "Sid": "AllowGetStateObject",
      "Effect": "Allow",
      "Action": "s3:GetObject",
      "Resource": "arn:aws:s3:::slack-audit-logs/slack/audit/state.json"
    }
  ]
}
    • Substitua slack-audit-logs se você tiver inserido um nome de bucket diferente.

  3. Clique em Próxima > Criar política.

  4. Acesse IAM > Funções > Criar função > Serviço da AWS > Lambda.

  5. Anexe a política recém-criada.

  6. Nomeie a função como SlackAuditToS3Role e clique em Criar função.

Criar a função Lambda

  1. No console da AWS, acesse Lambda > Functions > Create function.
  2. Clique em Criar do zero.
  3. Informe os seguintes detalhes de configuração:
Configuração Valor
Nome slack_audit_to_s3
Ambiente de execução Python 3.13
Arquitetura x86_64
Função de execução SlackAuditToS3Role

  1. Depois que a função for criada, abra a guia Código, exclua o stub e insira o seguinte (slack_audit_to_s3.py):

    #!/usr/bin/env python3
# Lambda: Pull Slack Audit Logs (Enterprise Grid) to S3 (no transform)

import os, json, time, urllib.parse
from urllib.request import Request, urlopen
from urllib.error import HTTPError, URLError
import boto3

BASE_URL = "https://api.slack.com/audit/v1/logs"

TOKEN        = os.environ["SLACK_AUDIT_TOKEN"]  # org-level user token with auditlogs:read
BUCKET       = os.environ["S3_BUCKET"]
PREFIX       = os.environ.get("S3_PREFIX", "slack/audit/")
STATE_KEY    = os.environ.get("STATE_KEY", "slack/audit/state.json")
LIMIT        = int(os.environ.get("LIMIT", "200"))             # Slack recommends <= 200
MAX_PAGES    = int(os.environ.get("MAX_PAGES", "20"))
LOOKBACK_SEC = int(os.environ.get("LOOKBACK_SECONDS", "3600")) # First-run window
HTTP_TIMEOUT = int(os.environ.get("HTTP_TIMEOUT", "60"))
HTTP_RETRIES = int(os.environ.get("HTTP_RETRIES", "3"))
RETRY_AFTER_DEFAULT = int(os.environ.get("RETRY_AFTER_DEFAULT", "2"))
# Optional server-side filters (comma-separated "action" values), empty means no filter
ACTIONS      = os.environ.get("ACTIONS", "").strip()

s3 = boto3.client("s3")

def _get_state() -> dict:
    try:
        obj = s3.get_object(Bucket=BUCKET, Key=STATE_KEY)
        st = json.loads(obj["Body"].read() or b"{}")
        return {"cursor": st.get("cursor")}
    except Exception:
        return {"cursor": None}

def _put_state(state: dict) -> None:
    body = json.dumps(state, separators=(",", ":")).encode("utf-8")
    s3.put_object(Bucket=BUCKET, Key=STATE_KEY, Body=body, ContentType="application/json")

def _http_get(params: dict) -> dict:
    qs  = urllib.parse.urlencode(params, doseq=True)
    url = f"{BASE_URL}?{qs}" if qs else BASE_URL
    req = Request(url, method="GET")
    req.add_header("Authorization", f"Bearer {TOKEN}")
    req.add_header("Accept", "application/json")

    attempt = 0
    while True:
        try:
            with urlopen(req, timeout=HTTP_TIMEOUT) as r:
                return json.loads(r.read().decode("utf-8"))
        except HTTPError as e:
            # Respect Retry-After on 429/5xx
            if e.code in (429, 500, 502, 503, 504) and attempt < HTTP_RETRIES:
                retry_after = 0
                try:
                    retry_after = int(e.headers.get("Retry-After", RETRY_AFTER_DEFAULT))
                except Exception:
                    retry_after = RETRY_AFTER_DEFAULT
                time.sleep(max(1, retry_after))
                attempt += 1
                continue
            # Re-raise other HTTP errors
            raise
        except URLError:
            if attempt < HTTP_RETRIES:
                time.sleep(RETRY_AFTER_DEFAULT)
                attempt += 1
                continue
            raise

def _write_page(payload: dict, page_idx: int) -> str:
    ts  = time.strftime("%Y/%m/%d/%H%M%S", time.gmtime())
    key = f"{PREFIX}/{ts}-slack-audit-p{page_idx:05d}.json"
    body = json.dumps(payload, separators=(",", ":")).encode("utf-8")
    s3.put_object(Bucket=BUCKET, Key=key, Body=body, ContentType="application/json")
    return key

def lambda_handler(event=None, context=None):
    state  = _get_state()
    cursor = state.get("cursor")

    params = {"limit": LIMIT}
    if ACTIONS:
        params["action"] = [a.strip() for a in ACTIONS.split(",") if a.strip()]
    if cursor:
        params["cursor"] = cursor
    else:
        # First run (or reset): fetch a recent window by time
        params["oldest"] = int(time.time()) - LOOKBACK_SEC

    pages = 0
    total = 0
    last_cursor = None

    while pages < MAX_PAGES:
        data = _http_get(params)
        _write_page(data, pages)

        entries = data.get("entries") or []
        total += len(entries)

        # Cursor for next page
        meta = data.get("response_metadata") or {}
        next_cursor = meta.get("next_cursor") or data.get("next_cursor")
        if next_cursor:
            params = {"limit": LIMIT, "cursor": next_cursor}
            if ACTIONS:
                params["action"] = [a.strip() for a in ACTIONS.split(",") if a.strip()]
            last_cursor = next_cursor
            pages += 1
            continue
        break

    if last_cursor:
        _put_state({"cursor": last_cursor})

    return {"ok": True, "pages": pages + (1 if total or last_cursor else 0), "entries": total, "cursor": last_cursor}

if __name__ == "__main__":
    print(lambda_handler())

  2. Acesse Configuração > Variáveis de ambiente > Editar > Adicionar nova variável de ambiente.

  3. Insira as seguintes variáveis de ambiente, substituindo pelos seus valores:

    Chave Valor de exemplo
    S3_BUCKET slack-audit-logs
    S3_PREFIX slack/audit/
    STATE_KEY slack/audit/state.json
    SLACK_AUDIT_TOKEN xoxp-*** (token de usuário no nível da organização com auditlogs:read)
    LIMIT 200
    MAX_PAGES 20
    LOOKBACK_SECONDS 3600
    HTTP_TIMEOUT 60
    HTTP_RETRIES 3
    RETRY_AFTER_DEFAULT 2
    ACTIONS (opcional, CSV) user_login,app_installed

  4. Depois que a função for criada, permaneça na página dela ou abra Lambda > Functions > sua-função.

  5. Selecione a guia Configuração.

  6. No painel Configuração geral, clique em Editar.

  7. Mude Tempo limite para 5 minutos (300 segundos) e clique em Salvar.

Criar uma programação do EventBridge

  1. Acesse Amazon EventBridge > Scheduler > Criar programação.
  2. Informe os seguintes detalhes de configuração:
    • Programação recorrente: Taxa (1 hour).
    • Destino: sua função Lambda slack_audit_to_s3.
    • Nome: slack-audit-1h.
  3. Clique em Criar programação.

Opcional: criar um usuário e chaves do IAM somente leitura para o Google SecOps

  1. No console da AWS, acesse IAM > Usuários > Adicionar usuários.
  2. Clique em Add users.
  3. Informe os seguintes detalhes de configuração:
    • Usuário: secops-reader.
    • Tipo de acesso: Chave de acesso — Acesso programático.
  4. Clique em Criar usuário.
  5. Anexe a política de leitura mínima (personalizada): Usuários > secops-reader > Permissões > Adicionar permissões > Anexar políticas diretamente > Criar política.

  6. No editor JSON, insira a seguinte política:

    {
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": ["s3:GetObject"],
      "Resource": "arn:aws:s3:::slack-audit-logs/*"
    },
    {
      "Effect": "Allow",
      "Action": ["s3:ListBucket"],
      "Resource": "arn:aws:s3:::slack-audit-logs"
    }
  ]
}

  7. Defina o nome como secops-reader-policy.

  8. Acesse Criar política > pesquise/selecione > Próxima > Adicionar permissões.

  9. Acesse Credenciais de segurança > Chaves de acesso > Criar chave de acesso.

  10. Faça o download do CSV (esses valores são inseridos no feed).

Configurar um feed no Google SecOps para ingerir registros de auditoria do Slack

  1. Acesse Configurações do SIEM > Feeds.
  2. Clique em + Adicionar novo feed.
  3. No campo Nome do feed, insira um nome para o feed (por exemplo, Slack Audit Logs).
  4. Selecione Amazon S3 V2 como o Tipo de origem.
  5. Selecione Auditoria do Slack como o Tipo de registro.
  6. Clique em Próxima.
  7. Especifique valores para os seguintes parâmetros de entrada:
    • URI do S3: s3://slack-audit-logs/slack/audit/
    • Opções de exclusão de fontes: selecione a opção de exclusão de acordo com sua preferência.
    • Idade máxima do arquivo: inclui arquivos modificados no último número de dias. O padrão é de 180 dias.
    • ID da chave de acesso: chave de acesso do usuário com acesso ao bucket do S3.
    • Chave de acesso secreta: chave secreta do usuário com acesso ao bucket do S3.
    • Namespace do recurso: o namespace do recurso.
    • Rótulos de ingestão: o rótulo aplicado aos eventos deste feed.
  8. Clique em Próxima.
  9. Revise a nova configuração do feed na tela Finalizar e clique em Enviar.

Tabela de mapeamento do UDM

Campo de registro Mapeamento do UDM Lógica
action metadata.product_event_type Mapeado diretamente do campo action no registro bruto.
actor.type principal.labels.value Mapeado diretamente do campo actor.type, com a chave actor.type adicionada.
actor.user.email principal.user.email_addresses Mapeado diretamente do campo actor.user.email.
actor.user.id principal.user.product_object_id Mapeado diretamente do campo actor.user.id.
actor.user.id principal.user.userid Mapeado diretamente do campo actor.user.id.
actor.user.name principal.user.user_display_name Mapeado diretamente do campo actor.user.name.
actor.user.team principal.user.group_identifiers Mapeado diretamente do campo actor.user.team.
context.ip_address principal.ip Mapeado diretamente do campo context.ip_address.
context.location.domain about.resource.attribute.labels.value Mapeado diretamente do campo context.location.domain, com a chave context.location.domain adicionada.
context.location.id about.resource.id Mapeado diretamente do campo context.location.id.
context.location.name about.resource.name Mapeado diretamente do campo context.location.name.
context.location.name about.resource.attribute.labels.value Mapeado diretamente do campo context.location.name, com a chave context.location.name adicionada.
context.location.type about.resource.resource_subtype Mapeado diretamente do campo context.location.type.
context.session_id network.session_id Mapeado diretamente do campo context.session_id.
context.ua network.http.user_agent Mapeado diretamente do campo context.ua.
context.ua network.http.parsed_user_agent Informações analisadas do user agent derivadas do campo context.ua usando o filtro parseduseragent.
country principal.location.country_or_region Mapeado diretamente do campo country.
date_create metadata.event_timestamp.seconds O carimbo de data/hora da época do campo date_create é convertido em um objeto de carimbo de data/hora.
details.inviter.email target.user.email_addresses Mapeado diretamente do campo details.inviter.email.
details.inviter.id target.user.product_object_id Mapeado diretamente do campo details.inviter.id.
details.inviter.name target.user.user_display_name Mapeado diretamente do campo details.inviter.name.
details.inviter.team target.user.group_identifiers Mapeado diretamente do campo details.inviter.team.
details.reason security_result.description Mapeado diretamente do campo details.reason ou, se for uma matriz, concatenado com vírgulas.
details.type about.resource.attribute.labels.value Mapeado diretamente do campo details.type, com a chave details.type adicionada.
details.type security_result.summary Mapeado diretamente do campo details.type.
entity.app.id target.resource.id Mapeado diretamente do campo entity.app.id.
entity.app.name target.resource.name Mapeado diretamente do campo entity.app.name.
entity.channel.id target.resource.id Mapeado diretamente do campo entity.channel.id.
entity.channel.name target.resource.name Mapeado diretamente do campo entity.channel.name.
entity.channel.privacy target.resource.attribute.labels.value Mapeado diretamente do campo entity.channel.privacy, com a chave entity.channel.privacy adicionada.
entity.file.filetype target.resource.attribute.labels.value Mapeado diretamente do campo entity.file.filetype, com a chave entity.file.filetype adicionada.
entity.file.id target.resource.id Mapeado diretamente do campo entity.file.id.
entity.file.name target.resource.name Mapeado diretamente do campo entity.file.name.
entity.file.title target.resource.attribute.labels.value Mapeado diretamente do campo entity.file.title, com a chave entity.file.title adicionada.
entity.huddle.date_end about.resource.attribute.labels.value Mapeado diretamente do campo entity.huddle.date_end, com a chave entity.huddle.date_end adicionada.
entity.huddle.date_start about.resource.attribute.labels.value Mapeado diretamente do campo entity.huddle.date_start, com a chave entity.huddle.date_start adicionada.
entity.huddle.id about.resource.attribute.labels.value Mapeado diretamente do campo entity.huddle.id, com a chave entity.huddle.id adicionada.
entity.huddle.participants.0 about.resource.attribute.labels.value Mapeado diretamente do campo entity.huddle.participants.0, com a chave entity.huddle.participants.0 adicionada.
entity.huddle.participants.1 about.resource.attribute.labels.value Mapeado diretamente do campo entity.huddle.participants.1, com a chave entity.huddle.participants.1 adicionada.
entity.type target.resource.resource_subtype Mapeado diretamente do campo entity.type.
entity.user.email target.user.email_addresses Mapeado diretamente do campo entity.user.email.
entity.user.id target.user.product_object_id Mapeado diretamente do campo entity.user.id.
entity.user.name target.user.user_display_name Mapeado diretamente do campo entity.user.name.
entity.user.team target.user.group_identifiers Mapeado diretamente do campo entity.user.team.
entity.workflow.id target.resource.id Mapeado diretamente do campo entity.workflow.id.
entity.workflow.name target.resource.name Mapeado diretamente do campo entity.workflow.name.
id metadata.product_log_id Mapeado diretamente do campo id.
ip principal.ip Mapeado diretamente do campo ip. Determinado por uma lógica baseada no campo action. O padrão é USER_COMMUNICATION, mas muda para outros valores, como USER_CREATION, USER_LOGIN, USER_LOGOUT, USER_RESOURCE_ACCESS, USER_RESOURCE_UPDATE_PERMISSIONS ou USER_CHANGE_PERMISSIONS, com base no valor de action. Fixado no código como "SLACK_AUDIT". Definido como "Enterprise Grid" se date_create existir. Caso contrário, definido como "Registros de auditoria" se user_id existir. Fixado no código como "Slack". Fixado no código como "REMOTE". Defina como "SSO" se action contiver "user_login" ou "user_logout". Caso contrário, defina como "MACHINE". Não mapeado nos exemplos fornecidos. O padrão é "ALLOW", mas é definido como "BLOCK" se action for "user_login_failed". Defina como "Slack" se date_create existir. Caso contrário, defina como "SLACK" se user_id existir.
user_agent network.http.user_agent Mapeado diretamente do campo user_agent.
user_id principal.user.product_object_id Mapeado diretamente do campo user_id.
username principal.user.product_object_id Mapeado diretamente do campo username.

Precisa de mais ajuda? Receba respostas de membros da comunidade e profissionais do Google SecOps.