Správa aktiv a tok činností

Instance Parsceku, který analyzuje událostní linku, může vysílat kompaktní události aktivit do vyhrazeného tématu Kafka. Služba LogMan.io Assets (lmio-assets) toto téma odebírá a aktualizuje inventář aktiv (poslední záznam, identity, zdrojové jméno v štítcích). Také odebírá složitou linku nájemce, aby udržovala časově degradované rizikové skóre entity pro každého principála. Viz LogMan.io Assets a Rizikové skóre entity.

To je oddělené od hlavního tématu událostí: parsované logy stále putují do kafka.events.topic na lince. Reportování aktivit používá jiné téma, které je určeno z deklarace linky událostí tenantů activity (/EventLanes/<tenant>/activity.yaml, výchozí název tématu events.<tenant>.activity).

Jak to funguje v Parsec

Při spuštění Parsec načítá /Schemas/Mappings/Asset_<schema>.yaml (například Asset_ECS.yaml pro ECS) a mapuje názvy polí na dva procesory v lmioparsec/processors/asset/:

1. AssetIdentityProcessor provádí kontrolu identity a volitelný záložní postup, poté vyplní context["asset"] při každé zpracované události.
2. IndicatorOfActivityProcessor (IOA, indikátor aktivity) čte tento kontext a vysílá události aktivity do Kafka tématu aktivity.

Oba procesory jsou připojeny na konci ParsecPipeline, po parsování, obohacení a kontrole typu, a před tím, než je událost zakódována a zapsána do výstupu events.

Sdílený ActivityPipeline je vytvořen jednou pro každou aplikaci Parsec. Procesor IOA posílá události aktivit do svého interního zdroje; pipeline kóduje JSON a zapisuje do Kafka tématu aktivit.

Mapování aktiv (Asset_<schema>.yaml)

Knihovní soubor s define.type: asset/mapping (zastaralá cesta Assets_<schema>.yaml je stále načítána jako záložní).

Mapování má tři části:

• Normalizace identity (identity + fallback): vyřešte host.id / user.id z polí události pomocí vyhledávání; na analyzovanou událost se zapíše nejvýše jedno ID principála.
• Identity pouze pro aktivitu (activity.identity + sdílený fallback): další pole se hlásí do streamu aktivit, aniž by se změnila analyzovaná událost (viz Pole identity pouze pro aktivitu).
• Payloady aktivit (mapping): mapujte vyřešené identity na názvy polí u aktivit v Kafka událostech.

Vlastní pole identity pouze pro aktivitu patří do activity.identity (mapování knihovny nebo sled událostí). Vlastní názvy polí payload patří do sekce mapping nebo přepsání sledů událostí.

Normalizace identity

Každý vyhledávací název uvádí pole událostí, jejichž hodnoty se používají jako klíče pro vyhledávání. Parsec volá odpovídající vyhledávací tabulku a z výsledku čte host.id nebo user.id.

/Schemas/Mappings/Asset_ECS.yaml

---
define:
  type: asset/mapping

identity:
  ips2hostid:
    - host.ip
  macs2hostid:
    - host.mac
  hostnames2hostid:
    - host.hostname
  usernames2userid:
    - user.name
  useremails2userid:
    - user.email

fallback:
  host.hostname: lowercase_and_domain_removal
  user.name: lowercase_and_domain_removal
  source.address: lowercase_and_domain_removal

activity:
  identity:
    hostnames2hostid:
      - host.hostname
      - source.address

mapping:
  tags: tags
  event_lane: observer.name
  observer_type: observer.type
  aux_hosts: related.hosts
  aux_ips: related.ip
  aux_macs: related.mac
  aux_users: related.user

Soubor knihovny ECS /Schemas/Mappings/Asset_ECS.yaml obsahuje toto uspořádání. host.hostname se objevuje jak v identity, tak v activity.identity; Parsec ho zpracovává jednou na hlavní cestě a přeskočí duplicitní výskyt v activity.identity. source.address je jen pro aktivitu a při jeho vyřešení pomocí vyhledávání nebo náhradní hodnoty produkuje další host.id na tématu aktivity.

Vyhledat jméno	Řeší	Typická pole událostí (ECS)
ips2hostid	host.id	host.ip
macs2hostid	host.id	host.mac
hostnames2hostid	host.id	host.hostname
usernames2userid	user.id	user.name
useremails2userid	user.id	user.email

Specifická pole pro událostní linku (například client.ip nebo dns.question.name) patří do konfigurace událostní linky, nikoli do globálního mapování.

Vyhledávací tabulky (hostnames2hostid, ips2hostid a další) jsou deklarovány v Knihovně pod /Lookups/. Podívejte se na Lookups pro obecné nastavení vyhledávání.

Záložní metody

Fallback se spouští pouze tehdy, když žádné pole pro vyhledání identity nevyřešilo host.id nebo user.id na události. Poté jsou vyzkoušeny nakonfigurované fallback metody na polích hostname / username (v pořadí polí).

Metoda	Účinek
lowercase_and_domain_removal	Odstraňte předponu 'DOMAIN\' malými písmeny; u hostitelských jmen také přidejte doménu z 'ips2domain', pokud název nemá tečku

Která pole dostanou záložní hodnotu: pouze pole uvedená pod fallback v mapování knihovny nebo event lane. Výchozí pokrytí mapování ECS zahrnuje host.hostname, user.name a source.address. Pole přidaná přes identity.<lookup>+ (například dns.question.name) běží pouze vyhledávání, dokud nepřidáte explicitní položku, například dns.question.name: lowercase_and_domain_removal.

Událostní sekce parsec.asset.fallback slučuje se s mapováním knihovny; nenahrazuje jej. Prazdné fallback: {} ponechává výchozí nastavení mapování beze změny.

Zakázání záložní možnosti pro pole, které již má výchozí mapování: nastavte false nebo null. Vyhledávání stále probíhá.

/EventLanes/mytenant/firewall.yaml

parsec:
  asset:
    fallback:
      host.hostname: false
      user.name: false

Neexistuje žádný globální přepínač pro okamžité vypnutí všech záloh.

Chcete-li přeskočit pouze hodnoty, které jsou samotnou IP adresou (a zachovat fallback pro skutečné doménová jména), použijte místo toho parsec.asset.skip.host.ip: true — viz Přeskočit IP adresy používané jako doménové názvy.

Mapování oblastí činnosti

Sekce mapping definuje názvy polí událostí používané v datech Kafka aktivit. Hlavní názvy polí (host.id, user.id, service.id) pocházejí ze schématu nájemce (define.principal_hostid, define.principal_userid, define.principal_serviceid; starší principal_host / principal_user / principal_service stále fungují).

Klíč	Výchozí (ECS)	Role
tags	tags	Technický seznam značek hlavních událostí
event_lane	observer.name	Řetězec cesty zdrojové / událostní stopy
observer_type	observer.type	Vždy nastavte na lmio-parsec při událostech aktivity
aux_hosts	related.hosts	Identita hostitele, když host.id chybí
aux_ips	related.ip	Seznam IP na událostech pomocného hostitele
aux_macs	related.mac	Seznam MAC na událostech pomocného hostitele
aux_users	related.user	Identita uživatele, když user.id chybí

/Schemas/Mappings/Asset_ECS.yaml

mapping:
  tags: tags
  event_lane: observer.name
  observer_type: observer.type
  aux_hosts: related.hosts
  aux_ips: related.ip
  aux_macs: related.mac
  aux_users: related.user

Konfigurace jízdního pruhu události (parsec.asset)

Přepsání pro jednotlivé linky pro vyhledávání identity, záložní variantu a hlášení aktivit se nachází pod parsec.asset v YAML souboru linky událostí. Hodnoty se slučují s /Schemas/Mappings/Asset_<schema>.yaml (včetně výchozího activity.identity na ECS).

/EventLanes/mytenant/firewall.yaml

---
define:
  type: lmio/event-lane

parsec:
  name: /Parsers/MyVendor/MyProduct/
  asset:
    # Activity Kafka flush (default: enabled, 60s unless overridden).
    heartbeat: 120s   # or ``off`` / ``false``

    # Drop host/user identities from the activity stream (case-insensitive).
    skip:
      host:
        exact:
          - localhost
        prefixes:
          - temp-
        suffixes:
          - .local
        contains:
          - -dhcp-
        ip: true   # host.hostname (or other host identity) is a bare IPv4/IPv6
      user:
        exact:
          - SYSTEM
        prefixes:
          - "NT AUTHORITY\\"
        suffixes:
          - "$"
        contains:
          - svc_
        ip: true

    identity:
      hostnames2hostid+:   # append to mapping list
        - dns.question.name
      ips2hostid+:         # append
        - client.ip
        - source.ip
      hostnames2hostid: []  # overwrite mapping list (empty = none from lane)

    fallback:
      dns.question.name: lowercase_and_domain_removal
      user.name: false

    # Extra fields for the activity stream only (never written on the parsed event).
    activity:
      identity:
        hostnames2hostid:
          - source.address
        usernames2userid:
          - source.user.name

Klíč	Výchozí	Účinek
heartbeat	povoleno, výchozí hodnota procesoru 60 s	Minimální počet sekund mezi flushi v Kafce; off zakazuje IndicatorOfActivityProcessor
skip.host / skip.user	(žádný)	exact (full string; names alias), prefixes, suffixes, contains, and/or ip: true to drop from activity (principals and related host/user) and to skip default fallback into host.id / user.id; case-insensitive
identity.<lookup>	(žádný)	Nahraďte seznam polí mapování pro tento vyhledávací dotaz
identity.<lookup>+	(žádný)	Přidejte pole do seznamu mapování
fallback.<field>	z mapování	Název záložní metody, nebo false / null pro deaktivaci pro tuto větev (použije se výchozí mapování, pokud je vynecháno)
activity.identity.<lookup>	(žádný)	Další pole událostí pro tok aktivit; pole již v identity (včetně výchozích mapování) jsou přeskočena
activity.identity.<lookup>+	(žádný)	Přidat pole do seznamu pouze pro vyhledávání aktivit

Komplementární filtr v Astatech

Událostní linka skip zastavuje identity před aktivitou Kafka topic. Chcete-li odstranit host/uživatelská jména v inventáři (včetně prefixů / sufixů / obsahuje / minimální počet pomlček / ip), použijte Knihovnu /Assets/Skip/ v nastavení Asetů.

Přeskočit IP adresy používané jako názvy hostitelů

Některé zdroje vloží samotnou IPv4 nebo IPv6 adresu do host.hostname (nebo jiného pole identity hostitele). Výchozí mapování ECS potom spustí záložní lowercase_and_domain_removal a tuto adresu zkopíruje do host.id.

Pro přeskočení pouze hostitelských identit ve tvaru IP a zachování záložní možnosti pro skutečné názvy hostitelů (FQDN, krátký název):

/EventLanes/mytenant/firewall.yaml

parsec:
  asset:
    skip:
      host:
        ip: true

Stejná vlajka existuje pro uživatele, když je user.name IP (skip.user.ip: true).

Efekt ip: true:

• fallback nekopíruje identitu do host.id / user.id
• Parsec nevysílá tuto identitu na Kafka téma aktivit
• vyhledávání názvů hostitele / uživatelských jmen stále probíhá (hostnames2hostid, usernames2userid)
• vyhledávání IP na speciálních polích IP (host.ip → ips2hostid) zůstává beze změny

Chcete-li zakázat všechny záložní hostnames na lince (nejen IP adresy), nastavte místo toho fallback.host.hostname: false — viz Metody zálohování.

Pole identity pouze pro aktivitu

Použijte parsec.asset.activity.identity, když protokol obsahuje více než jednu identitu hostitele nebo uživatele a chcete, aby každá z nich byla zaznamenána v inventáři aktiv, ale na samotné parsované události má být pouze jedno hlavní ID.

Typický příklad: host.hostname je pozorovaný host (server1) a source.address je jiný související host (server2). Výchozí nastavení ECS v Asset_ECS.yaml již oba uvádějí pod activity.identity.hostnames2hostid; host.hostname je přeskočeno při průchodu pouze aktivitami, protože je v identity.

Přepisování událostních pruhů se slučuje s knihovním mapováním stejným způsobem jako identity:

/EventLanes/mytenant/firewall.yaml

parsec:
  asset:
    activity:
      identity:
        hostnames2hostid+:
          - client.hostname

Parsec provádí stejná vyhledávání a záložní metody jako pro identity, ale:

• pole již uvedená v identity (včetně výchozích hodnot z Asset_<schema>.yaml) jsou přeskočena v activity.identity
• vyřešené hodnoty specifické pro aktivitu nejsou zapisovány do host.id / user.id v zpracované události
• každá zbývající vyřešená hodnota generuje další hlavní událost aktivity (host.id / user.id v payloadu aktivity Kafka)
• duplicitní ID již vyslaná z hlavní cesty identity jsou přeskočena jako druhé opatření

Pokud vyhledávání i náhradní možnost selžou, pole je hlášeno jako pomocná aktivita (related.hosts, related.user a tak dále) místo host.id.

Více hodnot host.id z jedné analyzované události

Když záznam obsahuje dvě identity hostitele (například host.hostname=server1 a source.address=server2), Parsec může vygenerovat dva hlavní události aktivity, zatímco analyzovaná událost si zachovává jedno host.id:

Krok	Zpracovaná událost	Téma aktivity Kafka
Hlavní identity na host.hostname	může nastavit host.id (první úspěšný vyhledávací pokus/náhradní řešení)	1× aktivita s tímto host.id
activity.identity on source.address	nezměněno	1× další aktivita s host.id pro server2

Obě zprávy o aktivitě obsahují @timestamp, tenant, observer.name, observer.type a značky Parsec. Jsou deduplikovány samostatně v IOA bufferu (klíčováno podle host.id:<hodnota>) a v rámci okna heartbeat.

Viz také Event Lanes pro obecnou konfiguraci eventové dráhy (Kafka témata, cesta parseru, schéma a časové pásmo).

Pracovní pruh události

Udalosti aktivit ze všech instancí Parsec pro nájemce sdílejí jedno téma Kafka. Název tématu pochází z deklarace kanálu událostí aktivity nájemce, obvykle /EventLanes/<tenant>/activity.yaml:

/EventLanes/mytenant/activity.yaml

---
define:
  type: lmio/event-lane

kafka:
  events:
    topic: events.mytenant.activity

Parsec řeší tuto cestu při spuštění (get_activity_kafka_topic). Výchozí název tématu je events.<tenant>.activity. Tento kanál není stejný jako zdrojová témata events.<tenant>.<stream>: analyzované logy stále jdou do kanálu deklarovaného na každé instanci Parsec; aktivita je samostatný kompaktní proud konzumovaný LogMan.io Assets.

Podívejte se na Kafka topics pro řádek aktivit v tabulce výchozích témat.

Co je extrahováno z každé analyzované události

AssetIdentityProcessor provádí vyhledávání z sloučených seznamů identit, poté volitelný záložní postup, a vyplňuje context["asset"]:

• Když je host.id / user.id znám z hlavní cesty identity: hlavní řádek kind: id pro aktivitu (maximálně jedno ID zapsané v analyzované události).
• Pole pod activity.identity přidávají další hlavní nebo pomocné řádky, aniž by měnily analyzovanou událost.
• Když hlavní ID není známo: pouze pomocné řádky (hosts s kind hostname | ip | mac; users s kind name | email, žádný řádek kind: id).

IndicatorOfActivityProcessor čte pouze tento kontext (plus služební účet z události):

• host_id / user_id přítomen → hlavní aktivita (schéma hlavních polí v Kafka payload)
• hostname name bez host_id → mapping.aux_hosts
• sirotčí ip / mac bez host_id → kombinovaná událost mapping.aux_ips / mapping.aux_macs
• name / email bez user_id → mapping.aux_users

Existující host.id / user.id na zpracované události

Pokud parser nebo obohacovač již nastavili pole principal, AssetIdentityProcessor jej nepřepíše. Vyhledávání se stále provádí a výsledky jsou uloženy do context["asset"]. Fallback se spustí pouze tehdy, když žádné vyhledávání nevyřešilo principal id a pole bylo prázdné.

Pořadí vyhledávání hostitele: nejprve pole hostnames2hostid, potom ips2hostid, poté macs2hostid; pokud všechny vyhledávání selžou, použijí se pole uvedená v fallback. Identity odpovídající skip.host nejsou zálohováním kopírovány do host.id.

Uživatel vyhledává objednávku: pole usernames2userid, poté useremails2userid; poté se použijí záložní nakonfigurovaná pole uživatelského jména. Identity odpovídající skip.user nejsou do user.id zkopírovány záložní cestou.

Tip

Odstraňte schéma enrich u polí hostname / username při používání těchto procesorů (stejné vyhledávání jako u DefaultLookupEnricher, ale zde sjednocené).

Seskupování a odstraňování duplicit

Události aktivit se hromadí v paměti, přičemž klíčem je identita (hlavní, pomocný host, pomocný uživatel). Při Application.tick/60!, po nakonfigurovaném intervalu heartbeat, jsou čekající události serializovány, přiřazeno jim deterministické _id a zapsány do Kafka.

Příklady dat aktivit

Hlavní činnost

{
  "host.id": "server01.example.com",
  "@timestamp": 1710000000000,
  "tenant": "mytenant",
  "tags": ["lmio-parsec:1.2.3", "instance-abc"],
  "observer.name": "/EventLanes/mytenant/fortinet.yaml",
  "observer.type": "lmio-parsec",
  "_id": 281474976710657
}

Druhý hlavní host z activity.identity

Stejná rozparsovaná událost jako výše, ale source.address vyřešil jiný hostitel (pole pouze pro aktivitu):

{
  "host.id": "server02.example.com",
  "@timestamp": 1710000000000,
  "tenant": "mytenant",
  "tags": ["lmio-parsec:1.2.3", "instance-abc"],
  "observer.name": "/EventLanes/mytenant/fortinet.yaml",
  "observer.type": "lmio-parsec",
  "_id": 281474976710660
}

Pomocná uživatelská aktivita

{
  "related.user": ["jdoe", "jdoe@example.com"],
  "@timestamp": 1710000000000,
  "tenant": "mytenant",
  "observer.name": "/EventLanes/mytenant/ad.yaml",
  "observer.type": "lmio-parsec",
  "_id": 281474976710658
}

Aktivita pomocného hostitele (IP a MAC, bez názvu hostitele)

{
  "related.ip": [{"h": 0, "l": 3232235777}],
  "related.mac": [11259375],
  "@timestamp": 1710000000000,
  "tenant": "mytenant",
  "observer.name": "/EventLanes/mytenant/dhcp.yaml",
  "observer.type": "lmio-parsec",
  "_id": 281474976710659
}

Pomocné hodnoty related.ip jsou v IndicatorOfActivityProcessor normalizovány na {h, l}. Hodnoty related.mac jsou celá čísla typu ui64.

Downstream: LogMan.io Assets

Aktivity zpráv Kafka jsou konzumovány LogMan.io Assets (lmio-assets), které provádějí upsert dokumentů Mongo s klíčem {tenant}:{kind}:{value} a aktualizují first_seen / last_seen.

Stejná služba se přihlašuje k nájemcovské komplexní událostní dráze (detekce korelátoru), aby spojila skóre rizika entity s exponenciálním poklesem. Toto chování je dokumentováno samostatně:

• LogMan.io Assets: přehled služby, pole aktiv, HTTP API
• Konfigurace aktiv: hostitelské a uživatelské filtry knihovny /Assets/Skip/
• Skóre rizika entity: model rozpadu a rozlišení od skóre rizika event.risk_score pro každé zjištění

Ohodnocení rizika detekce v korelátoru je popsáno v Ohodnocení rizika.