Coletar registros da plataforma Swimlane

Compatível com:

Google SecOps SIEM

Observação: esse recurso é coberto pelos Termos de produtos pré-GA dos Termos específicos de serviço do Google Security Operations. Os recursos pré-GA podem ter suporte limitado, e as mudanças realizadas podem não ser compatíveis com outras versões pré-GA. Para mais informações, consulte as diretrizes do serviço de suporte técnico do Google SecOps e os Termos específicos de serviço do Google SecOps.

Este documento explica como ingerir registros da plataforma Swimlane no Google Security Operations usando o Amazon S3.

Antes de começar

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

• Uma instância do Google SecOps
• Acesso privilegiado ao Swimlane (administrador da conta capaz de gerar um token de acesso pessoal)
• Acesso privilegiado à AWS (S3, IAM, Lambda, EventBridge)

Coletar os pré-requisitos da plataforma Swimlane (IDs, chaves de API, IDs da organização, tokens)

1. Faça login na plataforma Swimlane como um administrador da conta.
2. Acesse Opções de perfil.
3. Clique em Perfil para abrir o editor de perfil.
4. Navegue até a seção Token de acesso pessoal.
5. Clique em Gerar token para criar um novo token de acesso pessoal.
6. Copie o token imediatamente e armazene-o com segurança. Ele não será mostrado novamente.
7. Grave os seguintes detalhes da integração:
   • Token de acesso pessoal (PAT): usado no cabeçalho Private-Token para chamadas de API.
   • ID da conta: obrigatório para o caminho da API Audit Log /api/public/audit/account/{ACCOUNT_ID}/auditlogs. Entre em contato com o administrador do Swimlane se você não souber seu ID da conta.
   • URL base: seu domínio do Swimlane (por exemplo, https://eu.swimlane.app, https://us.swimlane.app).

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, swimlane-audit).
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": "AllowPutSwimlaneAuditObjects",
         "Effect": "Allow",
         "Action": ["s3:PutObject"],
         "Resource": "arn:aws:s3:::swimlane-audit/swimlane/audit/*"
       },
       {
         "Sid": "AllowStateReadWrite",
         "Effect": "Allow",
         "Action": ["s3:GetObject", "s3:PutObject"],
         "Resource": "arn:aws:s3:::swimlane-audit/swimlane/audit/state.json"
       }
     ]
   }
   

   • Substitua swimlane-audit 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 e a política gerenciada da AWS:

   • A política personalizada criada acima
   • service-role/AWSLambdaBasicExecutionRole (registros do CloudWatch)

6. Nomeie a função como WriteSwimlaneAuditToS3Role 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	swimlane_audit_to_s3
   Ambiente de execução	Python 3.13
   Arquitetura	x86_64
   Função de execução	WriteSwimlaneAuditToS3Role

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

   #!/usr/bin/env python3
   import os, json, gzip, io, uuid, datetime as dt, urllib.parse, urllib.request
   import boto3
   
   # ---- Environment ----
   S3_BUCKET = os.environ["S3_BUCKET"]
   S3_PREFIX = os.environ.get("S3_PREFIX", "swimlane/audit/")
   STATE_KEY = os.environ.get("STATE_KEY", S3_PREFIX + "state.json")
   BASE_URL = os.environ["SWIMLANE_BASE_URL"].rstrip("/")  # e.g., https://eu.swimlane.app
   ACCOUNT_ID = os.environ["SWIMLANE_ACCOUNT_ID"]
   TENANT_LIST = os.environ.get("SWIMLANE_TENANT_LIST", "")  # comma-separated; optional
   INCLUDE_ACCOUNT = os.environ.get("INCLUDE_ACCOUNT", "true").lower() == "true"
   PAGE_SIZE = int(os.environ.get("PAGE_SIZE", "100"))  # max 100
   WINDOW_MINUTES = int(os.environ.get("WINDOW_MINUTES", "15"))  # time range per run
   PAT_TOKEN = os.environ["SWIMLANE_PAT_TOKEN"]  # Personal Access Token
   TIMEOUT = int(os.environ.get("TIMEOUT", "30"))
   
   AUDIT_URL = f"{BASE_URL}/api/public/audit/account/{ACCOUNT_ID}/auditlogs"
   
   s3 = boto3.client("s3")
   
   # ---- Helpers ----
   
   def _http(req: urllib.request.Request):
       return urllib.request.urlopen(req, timeout=TIMEOUT)
   
   def _now():
       return dt.datetime.utcnow()
   
   def get_state() -> dict:
       try:
           obj = s3.get_object(Bucket=S3_BUCKET, Key=STATE_KEY)
           return json.loads(obj["Body"].read())
       except Exception:
           return {}
   
   def put_state(state: dict) -> None:
       state["updated_at"] = _now().isoformat() + "Z"
       s3.put_object(Bucket=S3_BUCKET, Key=STATE_KEY, Body=json.dumps(state).encode())
   
   def build_url(from_dt: dt.datetime, to_dt: dt.datetime, page: int) -> str:
       params = {
           "pageNumber": str(page),
           "pageSize": str(PAGE_SIZE),
           "includeAccount": str(INCLUDE_ACCOUNT).lower(),
           "fromdate": from_dt.replace(microsecond=0).isoformat() + "Z",
           "todate": to_dt.replace(microsecond=0).isoformat() + "Z",
       }
       if TENANT_LIST:
           params["tenantList"] = TENANT_LIST
       return AUDIT_URL + "?" + urllib.parse.urlencode(params)
   
   def fetch_page(url: str) -> dict:
       headers = {
           "Accept": "application/json",
           "Private-Token": PAT_TOKEN,
       }
       req = urllib.request.Request(url, headers=headers)
       with _http(req) as r:
           return json.loads(r.read())
   
   def write_chunk(items: list[dict], ts: dt.datetime) -> str:
       key = f"{S3_PREFIX}{ts:%Y/%m/%d}/swimlane-audit-{uuid.uuid4()}.json.gz"
       buf = io.BytesIO()
       with gzip.GzipFile(fileobj=buf, mode="w") as gz:
           for rec in items:
               gz.write((json.dumps(rec) + "n").encode())
       buf.seek(0)
       s3.upload_fileobj(buf, S3_BUCKET, key)
       return key
   
   def lambda_handler(event=None, context=None):
       state = get_state()
   
       # determine window
       to_dt = _now()
       from_dt = to_dt - dt.timedelta(minutes=WINDOW_MINUTES)
       if (prev := state.get("last_to_dt")):
           try:
               from_dt = dt.datetime.fromisoformat(prev.replace("Z", "+00:00"))
           except Exception:
               pass
   
       page = int(state.get("page", 1))
       total_written = 0
   
       while True:
           url = build_url(from_dt, to_dt, page)
           resp = fetch_page(url)
           items = resp.get("auditlogs", []) or []
           if items:
               write_chunk(items, _now())
               total_written += len(items)
           next_path = resp.get("next")
           if not next_path:
               break
           page += 1
           state["page"] = page
   
       # advance state window
       state["last_to_dt"] = to_dt.replace(microsecond=0).isoformat() + "Z"
       state["page"] = 1
       put_state(state)
   
       return {"ok": True, "written": total_written, "from": from_dt.isoformat() + "Z", "to": to_dt.isoformat() + "Z"}
   
   if __name__ == "__main__":
       print(lambda_handler())
   

5. Acesse Configuração > Variáveis de ambiente.

6. Clique em Editar > Adicionar nova variável de ambiente.

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

   Chave	Valor de exemplo
   S3_BUCKET	swimlane-audit
   S3_PREFIX	swimlane/audit/
   STATE_KEY	swimlane/audit/state.json
   SWIMLANE_BASE_URL	https://eu.swimlane.app
   SWIMLANE_ACCOUNT_ID	xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx
   SWIMLANE_TENANT_LIST	tenantA,tenantB (opcional)
   INCLUDE_ACCOUNT	true
   PAGE_SIZE	100
   WINDOW_MINUTES	15
   SWIMLANE_PAT_TOKEN	<your-personal-access-token>
   TIMEOUT	30

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

9. Selecione a guia Configuração.

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

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

    Observação: o tempo limite é um limite máximo fixo. Isso significa que, se a função ainda estiver em execução quando o timer atingir 5 minutos, o AWS Lambda vai encerrá-la. Execuções mais curtas terminam normalmente, e você só recebe cobranças pelo tempo de execução real.

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 (15 min)
   • Destino: sua função Lambda swimlane_audit_to_s3
   • Nome: swimlane-audit-schedule-15min
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:::swimlane-audit/*"
       },
       {
         "Effect": "Allow",
         "Action": ["s3:ListBucket"],
         "Resource": "arn:aws:s3:::swimlane-audit"
       }
     ]
   }
   

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 da plataforma Swimlane

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, Swimlane Platform logs).
4. Selecione Amazon S3 V2 como o Tipo de origem.
5. Selecione Plataforma de faixa como o Tipo de registro.
6. Clique em Próxima.
7. Especifique valores para os seguintes parâmetros de entrada:
   • URI do S3: s3://swimlane-audit/swimlane/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. 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.

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