Rozpoczęcie pracy z grafami niestandardowymi w Microsoft Sentinel (wersja zapoznawcza)

Wykresy niestandardowe w Microsoft Sentinel umożliwiają badaczom zabezpieczeń i analitykom tworzenie dostosowanych reprezentacji grafu danych zabezpieczeń. Tworząc wykresy niestandardowe, można modelować określone wzorce ataków, badać zagrożenia i uruchamiać zaawansowane algorytmy grafów, aby odkryć ukryte relacje w środowisku cyfrowym. W tym przewodniku przedstawiono kroki tworzenia wykresów niestandardowych i zarządzania nimi przy użyciu notesów Jupyter w rozszerzeniu Microsoft Sentinel Visual Studio Code.

Ten artykuł koncentruje się na ręcznym tworzeniu wykresów niestandardowych przy użyciu kodu. Aby zapoznać się ze środowiskiem opartym na sztucznej inteligencji, zobacz Tworzenie wykresów niestandardowych wspomaganych przez sztuczną inteligencję w Microsoft Sentinel.

Wymagania wstępne

Do utworzenia wykresów niestandardowych w Microsoft Sentinel są wymagane następujące elementy:

• rozszerzenie Microsoft Sentinel dla Visual Studio Code. Aby uzyskać więcej informacji, zobacz Uruchamianie notatników w jeziorze danych Microsoft Sentinel.
• Rozszerzenie Jupyter dla Visual Studio Code.
• Microsoft Sentinel data lake skonfigurowane z odpowiednimi uprawnieniami. Aby uzyskać więcej informacji, zobacz Dołączanie do Microsoft Sentinel data lake.

Włącz łącznik Microsoft Entra ID, aby pozyskiwać tabele zasobów Microsoft Entra używane w przykładowym kodzie tego artykułu. Aby uzyskać więcej informacji, zobacz Pozyskiwanie danych zasobów w Microsoft Sentinel data lake.

Uprawnienia

Do interakcji z wykresami niestandardowymi potrzebne są następujące uprawnienia XDR w Sentinel data lake. W poniższej tabeli wymieniono wymagania dotyczące uprawnień dla typowych operacji grafów:

Role Microsoft Entra ID zapewniają szeroki dostęp do wszystkich obszarów roboczych w jeziorze danych. Aby uzyskać więcej informacji, zobacz Role i uprawnienia w Microsoft Sentinel.

Instalowanie Visual Studio Code i rozszerzenia Microsoft Sentinel

Tworzenie wykresów niestandardowych przy użyciu notesów Jupyter w rozszerzeniu Microsoft Sentinel Visual Studio Code. Aby uzyskać więcej informacji, zobacz Instalowanie Visual Studio Code i rozszerzenia Microsoft Sentinel

Tworzenie grafu niestandardowego

Aby utworzyć wykresy niestandardowe i pracować z nie, wykonaj następujące kroki:

1. Modelowanie wykresu niestandardowego
2. Zachowanie niestandardowego grafu przez zaplanowanie zadania dla grafu
3. Wyświetlanie wykresów niestandardowych i zarządzanie nimi

Modelowanie wykresu niestandardowego

Utwórz wykres niestandardowy przy użyciu notesu Jupyter w rozszerzeniu Microsoft Sentinel Visual Studio Code.

Poniższe kroki przeprowadzą Cię przez proces tworzenia pierwszego wykresu niestandardowego przy użyciu przykładowego notesu.

Skonfiguruj notatnik i połącz go z jeziorem danych

Wykonaj poniższe kroki, aby utworzyć notes i nawiązać połączenie z usługą data lake Microsoft Sentinel.

1. W Visual Studio Code z zainstalowanym rozszerzeniem Microsoft Sentinel wybierz ikonę Microsoft Sentinel w menu po lewej stronie.

2. Wybierz pozycję Zaloguj się, aby wyświetlić wykresy

3. Zostanie wyświetlone okno dialogowe z tekstem Rozszerzenie "Microsoft Sentinel" chce się zalogować przy użyciu firmy Microsoft. Wybierz pozycję Zezwalaj , aby się zalogować.

4. Zaloguj się przy użyciu poświadczeń.

5. Po zalogowaniu wybierz +, a następnie wybierz Utwórz nowy notatnik.

6. Nadaj plikowi notesu nazwę i zapisz go w odpowiedniej lokalizacji w obszarze roboczym.

   

7. Wybierz Wybierz jądro w prawym górnym rogu okna notatnika, aby wybrać pulę obliczeniową Spark.

8. Wybierz Microsoft Sentinel, a następnie wybierz dowolną z dostępnych pul platformy Spark.

   

   Wskazówka

   Możesz użyć promptów AI, aby utworzyć niestandardowy notatnik grafowy. Aby uzyskać więcej informacji, zobacz Tworzenie grafu niestandardowego wspomaganego przez sztuczną inteligencję w Microsoft Sentinel.

9. Aby uruchomić komórkę, wybierz ikonę trójkąta uruchamiania po lewej stronie komórki. Po pierwszym uruchomieniu komórki może zostać wyświetlony monit o wybranie jądra, jeśli jeszcze go nie wybrano.

   Uruchomienie komórki przy pierwszym uruchomieniu sesji platformy Spark trwa około pięciu minut.

   

Tworzenie grafu

Poniższy przykład tworzy graf do przechodzenia członkostw w grupach w usłudze Microsoft Entra i zrozumienia zagnieżdżonych relacji między grupami. Przykładowy kod pomaga zacząć od prostego scenariusza użycia, aby poznać możliwości grafu niestandardowego i wykorzystać możliwości przeszukiwania grafu w swoich analizach. Wykres można utworzyć na podstawie dowolnej tabeli dostępnej w Sentinel data lake.

1. Połącz się z obszarem roboczym i odczytaj tabele zasobów Entra, aby rozpocząć tworzenie grafu.

   from pyspark.sql import functions as F
   from sentinel_lake.providers import MicrosoftSentinelProvider
   
   lake_provider = MicrosoftSentinelProvider(spark=spark)
   
   # Use the "System tables" workspace which contains the Entra* Assets tables
   # If you are data is in a different workspace, update this variable accordingly and ensure the tables are present
   LOG_ANALYTICS_WORKSPACE = "System tables"
   
   # Dynamically get the latest snapshot time from EntraUsers
   snapshot_time = (
       lake_provider.read_table("EntraUsers", LOG_ANALYTICS_WORKSPACE)
       .df.agg(F.max("_SnapshotTime").alias("max_snapshot"))
       .collect()[0]["max_snapshot"]
       .strftime("%Y-%m-%dT%H:%M:%SZ")
   )
   print(f"Using snapshot_time: {snapshot_time}")
   
   snapshot_filter = (F.col("_SnapshotTime") == F.lit(snapshot_time).cast("timestamp"))
   
   # Load EntraMembers - edges: group contains user/group/servicePrincipal
   df_members = (
       lake_provider.read_table("EntraMembers", LOG_ANALYTICS_WORKSPACE)
       .filter(
           snapshot_filter
           & (F.col("sourceType") == "group")
           & (F.col("targetType").isin("user", "group", "servicePrincipal"))
       )
   )
   
   # Load EntraGroups - nodes
   df_groups = (
       lake_provider.read_table("EntraGroups", LOG_ANALYTICS_WORKSPACE)
       .filter(snapshot_filter)
       .select("id", "displayName", "mailEnabled")
   )
   
   # Load EntraUsers - nodes
   df_users = (
       lake_provider.read_table("EntraUsers", LOG_ANALYTICS_WORKSPACE)
       .filter(snapshot_filter)
       .select("id", "accountEnabled", "displayName", "department",
               "lastPasswordChangeDateTime", "userPrincipalName", "usageLocation")
   )
   
   # Load EntraServicePrincipals - nodes
   df_service_principals = (
       lake_provider.read_table("EntraServicePrincipals", LOG_ANALYTICS_WORKSPACE)
       .filter(snapshot_filter)
       .select("accountEnabled", "id", "displayName", "servicePrincipalType",
               "tenantId", "organizationId")
   )
   
   # Fix for Spark 3.x Parquet datetime rebase issue. Required when reading Parquet files
   # written by Spark 2.x which used the Julian calendar, whereas Spark 3.x uses Proleptic
   # Gregorian. Without these settings, timestamp columns (e.g. lastPasswordChangeDateTime)
   # may throw errors or return incorrect values. Safe to remove if all data was written by
   # Spark 3.x (typical for current Fabric/Sentinel environments).
   spark.conf.set("spark.sql.parquet.datetimeRebaseModeInRead", "CORRECTED")
   spark.conf.set("spark.sql.parquet.datetimeRebaseModeInWrite", "CORRECTED")
   spark.conf.set("spark.sql.parquet.int96RebaseModeInRead", "CORRECTED")
   spark.conf.set("spark.sql.parquet.int96RebaseModeInWrite", "CORRECTED")
   

2. Przygotowywanie ramek danych węzła i krawędzi wymaganych do utworzenia grafu

   # ============================================================
   # NODE PREPARATION
   # ============================================================
   
   # EntraUser nodes - keyed by user id
   user_nodes = (
       df_users.df
       .select(
           F.col("id"),
           F.col("displayName"),
           F.col("accountEnabled"),
           F.col("department"),
           F.col("lastPasswordChangeDateTime"),
           F.col("userPrincipalName"),
           F.col("usageLocation")
       )
   )
   
   # EntraGroup nodes - keyed by group id
   group_nodes = (
       df_groups.df
       .select(
           F.col("id"),
           F.col("displayName"),
           F.col("mailEnabled")
       )
   )
   
   # EntraServicePrincipal nodes - keyed by SP id
   sp_nodes = (
       df_service_principals.df
       .select(
           F.col("id"),
           F.col("displayName"),
           F.col("accountEnabled"),
           F.col("servicePrincipalType"),
           F.col("tenantId"),
           F.col("organizationId")
       )
   )
   
   # ============================================================
   # EDGE PREPARATION
   # ============================================================
   
   # Edge: EntraGroup --Contains--> EntraUser
   edge_group_contains_user = (
       df_members.df
       .filter(F.col("targetType") == "user")
       .select(
           F.col("sourceId").alias("SourceGroupId"),
           F.col("targetId").alias("TargetUserId")
       )
       .distinct()
       .withColumn("EdgeKey", F.concat_ws("_", F.col("SourceGroupId"), F.col("TargetUserId")))
   )
   
   # Edge: EntraGroup --Contains--> EntraGroup (nested groups)
   edge_group_contains_group = (
       df_members.df
       .filter(F.col("targetType") == "group")
       .select(
           F.col("sourceId").alias("SourceGroupId"),
           F.col("targetId").alias("TargetGroupId")
       )
       .distinct()
       .withColumn("EdgeKey", F.concat_ws("_", F.col("SourceGroupId"), F.col("TargetGroupId")))
   )
   
   # Edge: EntraGroup --Contains--> EntraServicePrincipal
   edge_group_contains_sp = (
       df_members.df
       .filter(F.col("targetType") == "servicePrincipal")
       .select(
           F.col("sourceId").alias("SourceGroupId"),
           F.col("targetId").alias("TargetSPId")
       )
       .distinct()
       .withColumn("EdgeKey", F.concat_ws("_", F.col("SourceGroupId"), F.col("TargetSPId")))
   )
   

3. Zdefiniuj schemat grafu i powiąż go z ramkami danych węzłów i krawędzi przygotowanymi wcześniej w tej procedurze

   from sentinel_graph import GraphSpecBuilder, Graph
   
   # Define the graph schema 
   
   entra_group_graph_spec = (
       GraphSpecBuilder.start()
   
       # === NODES ===
   
       .add_node("EntraUser")
       .from_dataframe(user_nodes)  # Native Spark DataFrame (from .df + .select + .distinct)
       .with_columns(
           "id", "displayName", "accountEnabled",
           "department", "lastPasswordChangeDateTime", "userPrincipalName", "usageLocation",
           key="id", display="displayName"
       )
   
       .add_node("EntraGroup")
       .from_dataframe(group_nodes)  # Native Spark DataFrame
       .with_columns(
           "id", "displayName", "mailEnabled",
           key="id", display="displayName"
       )
   
       .add_node("EntraServicePrincipal")
       .from_dataframe(sp_nodes)  # Native Spark DataFrame
       .with_columns(
           "id", "displayName", "accountEnabled",
           "servicePrincipalType", "tenantId", "organizationId",
           key="id", display="displayName"
       )
   
       # === EDGES ===
   
       # EntraGroup --ContainsUser--> EntraUser
       .add_edge("ContainsUser")
       .from_dataframe(edge_group_contains_user)  # Native Spark DataFrame
       .source(id_column="SourceGroupId", node_type="EntraGroup")
       .target(id_column="TargetUserId", node_type="EntraUser")
       .with_columns("SourceGroupId", "TargetUserId", "EdgeKey",
                     key="EdgeKey", display="EdgeKey")
   
       # EntraGroup --ContainsGroup--> EntraGroup (nested groups)
       .add_edge("ContainsGroup")
       .from_dataframe(edge_group_contains_group)  # Native Spark DataFrame
       .source(id_column="SourceGroupId", node_type="EntraGroup")
       .target(id_column="TargetGroupId", node_type="EntraGroup")
       .with_columns("SourceGroupId", "TargetGroupId", "EdgeKey",
                     key="EdgeKey", display="EdgeKey")
   
       # EntraGroup --ContainsServicePrincipal--> EntraServicePrincipal
       .add_edge("ContainsServicePrincipal")
       .from_dataframe(edge_group_contains_sp)  # Native Spark DataFrame
       .source(id_column="SourceGroupId", node_type="EntraGroup")
       .target(id_column="TargetSPId", node_type="EntraServicePrincipal")
       .with_columns("SourceGroupId", "TargetSPId", "EdgeKey",
                     key="EdgeKey", display="EdgeKey")
   
   ).done()
   

4. Weryfikowanie schematu grafu

   # Check the schema of the graph spec to ensure it's correct
   entra_group_graph_spec.show_schema()
   

5. Tworzenie grafu, w tym przygotowywanie danych i publikowanie grafu

   # Build the graph from the spec - this will validate the spec and prepare it for querying
   # Alter options is to use Graph.prepare() to prepare the graph nodes and edges in the lake
   # and then use Graph.publish() to create the graph. You would typically call prepare() and publish()
   # seperately to understand the cost of Graph API calls that are triggeterd by Graph.publish()
   # see https://learn.microsoft.com/azure/sentinel/billing?tabs=simplified%2Ccommitment-tiers
   intra_group_graph = Graph.build(entra_group_graph_spec)
   

   Uwaga

   Wykresy utworzone podczas interakcyjnych sesji notatnika są usuwane po zamknięciu sesji notatnika. Aby utrwalić graf do ponownego użycia i udostępniania, zobacz Utrwalanie grafu niestandardowego

Utworzono teraz wykres w notatniku.

Poniższy przykład Python uruchamia zapytanie Graph Query Language (GQL) na właśnie utworzonym przez Ciebie grafie, aby zbadać zagnieżdżone relacje grupowe. Używaj zapytań takich jak to po utworzeniu grafu, aby przechodzić przez hierarchie członkostwa i odkrywać pośrednie ścieżki dostępu. Aby spróbować, wklej kod do nowej komórki notatnika i uruchom go. To przykładowe zapytanie przechodzi przez zagnieżdżone członkostwa grup Microsoft Entra do ośmiu poziomów i wyświetla powstałe relacje:

# Query 1: Find nested group relationships nexting up to 8 levels deep
# Update the Entra Group name that you want to traverse from
query_nested_groups = """
MATCH p=(g1:EntraGroup)-[cg]->{1,8}(g2)
WHERE g1.displayName = 'tmplevel3'
RETURN *
"""
intra_group_graph.query(query_nested_groups).show()

Ten kod uruchamia przykładowe zapytanie języka Graph Query Language (GQL), aby pobrać wszystkie zagnieżdżone członkostwo w grupach o głębokości do 8 poziomów Wynikowy wykres jest wizualizowany w danych wyjściowych

Utrwalanie wykresu niestandardowego

Po utworzeniu kodu grafu w notesie możesz uruchomić notes w sesji interaktywnej lub zaplanować zadanie grafu. Wykresy utworzone podczas sesji notesu interakcyjnego są tymczasowe i są dostępne tylko w kontekście sesji notesu. Aby zapisać graf i udostępnić go zespołowi, zaplanuj zadanie grafu, aby często odbudowywać graf. Po zapisaniu grafu jest on dostępny z poziomu: środowiska grafu w portalu Microsoft Defender w obszarze Sentinel, notesów Visual Studio Code i interfejsów API zapytań programu Graph.

1. W notesie grafu wybierz pozycję Utwórz zaplanowane zadanie, a następnie wybierz pozycję Utwórz zadanie grafu.

   

2. W formularzu Utwórz zadanie grafu wprowadź Nazwę grafu i Opis, a następnie sprawdź, czy w polu Ścieżka uwzględniono właściwy notatnik grafu.

3. Aby utworzyć wykres bez konfigurowania harmonogramu odświeżania, wybierz pozycję Na żądanie w sekcji Harmonogram , a następnie wybierz pozycję Prześlij , aby utworzyć wykres.

   Uwaga

   Wykresy utworzone przy użyciu harmonogramu na żądanie mają domyślny okres przechowywania wynoszący 30 dni i są usuwane po wygaśnięciu.

4. Aby utworzyć wykres, na którym dane grafu są regularnie odświeżane, wybierz pozycję Zaplanowane w sekcji Harmonogram .

   1. Wybierz częstotliwość powtarzania dla zadania. Możesz wybrać opcję Według minuty, godziny, tygodniowego, dziennego lub miesięcznego.

   2. Zostanie wyświetlonych więcej opcji konfigurowania harmonogramu w zależności od wybranej częstotliwości. Na przykład dzień tygodnia, godzina dnia lub dzień miesiąca.

   3. Wybierz godzinę Rozpocznij o, od której harmonogram ma zacząć działać.

   4. Wybierz Czas zakończenia, aby harmonogram przestał być uruchamiany. Jeśli nie chcesz ustawiać godziny zakończenia harmonogramu, wybierz pozycję Ustaw zadanie tak, aby było uruchamiane przez czas nieokreślony. Daty i godziny znajdują się w strefie czasowej.

   5. Wybierz pozycję Prześlij , aby zapisać konfigurację zadania i opublikować zadanie. Proces tworzenia grafu rozpoczyna się w Twojej organizacji. Wyświetl nowo utworzony wykres i jego najnowszy stan w rozszerzeniu Sentinel.

   

Wyświetlanie wykresów niestandardowych i zarządzanie nimi

Po utworzeniu zadania grafowego możesz wyświetlać graf i zarządzać nim w swojej dzierżawie za pomocą rozszerzenia Microsoft Sentinel w programie Visual Studio Code.

1. Z listy wykresów wybierz zmaterializowany wykres, aby wyświetlić jego szczegóły.

2. Wybierz kartę Szczegóły zadania , aby wyświetlić stan zadania grafu, w tym czas ostatniego uruchomienia, czas następnego uruchomienia i wszelkie błędy napotkane podczas procesu kompilacji.

3. Wybierz pozycję Uruchom teraz, aby ręcznie wyzwolić kompilację grafu poza zaplanowanym czasem. Status zmienia się na W kolejce, a następnie na „W toku” podczas tworzenia wykresu.

   

4. Po zakończeniu kompilacji grafu stan zostanie zaktualizowany do pozycji Gotowe. Wybierz kartę Szczegóły grafu , aby wyświetlić informacje o grafie.

   

5. Teraz możesz wykonywać zapytania i wizualizować graf, korzystając z wizualizacji grafu w Microsoft Sentinel w portalu Defender. Aby uzyskać więcej informacji, zobacz Visualize graphs in Microsoft Sentinel graph (wersja zapoznawcza).

Aktualizowanie wykresów niestandardowych

Po zaktualizowaniu kodu grafu w notesie uruchom notes lokalnie, aby zweryfikować zmiany przed zaktualizowaniem zadania grafu.

Edytuj szczegóły istniejącego niestandardowego wykresu

Użyj tego przepływu pracy, aby edytować istniejące zadanie wykresu.

1. W panelu Microsoft Sentinel Graph w programie Visual Studio Code wybierz pozycję Edytuj zadanie grafu dla grafu, który chcesz zaktualizować, a następnie przejdź do pozycji Nazwa grafu>Szczegóły zadania.

2. Zaktualizuj pola, które chcesz zmienić:

   • Nazwa grafu (opcjonalnie). Zaktualizowanie nazwy grafu spowoduje utworzenie nowego grafu w dzierżawcy. Istniejący graf pozostaje niezmieniony.
   • Opis (opcjonalnie). Aktualizacja opisu zastępuje bieżący opis grafu i ponownie kompiluje graf.
   • Konfiguracja klastra (opcjonalnie). Aktualizowanie konfiguracji klastra zmienia rozmiar obliczeniowy i ponownie kompiluje graf.
   • Częstotliwość zadania (opcjonalnie). Aktualizowanie częstotliwości zadań zmienia harmonogram cykliczny i ponownie kompiluje graf.

Nadpisz istniejący niestandardowy wykres

Użyj tego przepływu pracy, gdy chcesz ponownie utworzyć graf i zastąpić istniejący graf o tej samej nazwie.

1. Otwórz zaktualizowany notatnik.

2. Wybierz Utwórz zaplanowane zadanie w środowisku notesu w usłudze Sentinel, a następnie wybierz Utwórz zadanie wykresu.

3. W formularzu Tworzenie grafu podaj nazwę grafu .

   • Jeśli podasz nową nazwę grafu, nowy graf zostanie utworzony w dzierżawcy, a istniejący graf pozostanie dostępny.
   • Jeśli ponownie użyjesz istniejącej nazwy wykresu dla dzierżawcy, istniejący wykres zostanie nadpisany po potwierdzeniu. Nazwy grafów są unikatowe, zduplikowane nazwy grafu nie są obsługiwane.

4. Sprawdź, czy Path odwołuje się do zaktualizowanego notatnika.

5. Zaktualizuj opis grafu.

6. Wybierz pozycję Konfiguracja klastra i Częstotliwość zadań, podobnie jak w przypadku tworzenia nowego grafu.

7. W razie potrzeby zaktualizuj informacje o harmonogramie .

8. Zaktualizuj ponownie pole Opis zgodnie z potrzebami.

9. Wybierz pozycję Prześlij, aby odtworzyć graf w swojej dzierżawie.

Treści powiązane

• Tworzenie grafu niestandardowego wspomaganego przez sztuczną inteligencję w Microsoft Sentinel
• Dokumentacja referencyjna biblioteki dostawcy grafów Microsoft Sentinel
• Informacje referencyjne dotyczące języka Graph Query Language (GQL) dla grafu niestandardowego w usłudze Sentinel
• Wizualizowanie wykresów na wykresie Microsoft Sentinel (wersja zapoznawcza)