In questo tutorial, costruiamo un workflow avanzato utilizzando PyGraphistry per analisi grafiche interattive nel contesto dell'analisi della sicurezza e dell'investigazione dei rischi. Iniziamo generando un dataset aziendale realistico di accesso, trasformandolo in nodi e archi, e arricchiamo il grafo con punteggi di rischio, indicatori di anomalia, metriche di centralità, rilevamento di comunità e incapsulamento di layout. Successivamente, utilizziamo PyGraphistry per collegare la struttura del grafo ad encodings visivi, etichette, tooltip e sottografi filtrati, generando visualizzazioni interattive locali in assenza di credenziali Graphistry.
Il tutorial dimostra come l'intelligenza grafica permetta di indagare utenti sospetti, dispositivi rischiosi, relazioni IP, servizi sensibili e schemi comportamentali ad alto rischio in un ambiente di analisi della sicurezza pratico.
Installa PyGraphistry e Dipendenze
import os, sys, subprocess, warnings, textwrap, json, math, random
warnings.filterwarnings("ignore")
def pip_install(packages):
subprocess.run([sys.executable, "-m", "pip", "install", "-q", "-U", *packages], check=True)
pip_install([
"graphistry[networkx,umap-learn]",
"pandas",
"numpy",
"networkx",
"scikit-learn",
"pyvis",
"matplotlib",
"pyarrow"
])
import numpy as np
import pandas as pd
import networkx as nx
import matplotlib.pyplot as plt
import graphistry
from pathlib import Path
from IPython.display import display, HTML, IFrame
from sklearn.preprocessing import StandardScaler
from sklearn.ensemble import IsolationForest
from sklearn.decomposition import PCA
from pyvis.network import Network
Creiamo l'ambiente completo per Colab installando PyGraphistry e tutte le librerie supportive necessarie per l'analisi grafica, la visualizzazione e il machine learning. Configuriamo la directory di output, il seme casuale e le credenziali Graphistry, affinché il notebook funzioni sia in locale che con Graphistry Hub. Inoltre, definiamo un'aiuto per la denominazione dei nodi per mantenere chiaramente separato ogni tipo di entità nel grafo.
Generare il Dataset di Accesso Aziendale
Per rappresentare i dati aziendali di accesso, generiamo un set realistico di informazioni che include utenti, dispositivi, indirizzi IP, servizi, ruoli e geografie.
Configurazione Parametri
- nusers: 55
- ndevices: 42
- nips: 36
- nservices: 15
- nroles: 7
- ngeos: 10
Esempio di Codice:
n_users = 55
n_devices = 42
n_ips = 36
n_services = 15
n_roles = 7
n_geos = 10
n_events = 2200
users = [f"user{i:03d}" for i in range(nusers)]
devices = [f"device{i:03d}" for i in range(ndevices)]
ips = [f"10.{i // 255}.{i % 255}.{rng.integers(1, 255)}" for i in range(1, n_ips + 1)]
services = [
"salesforce", "snowflake", "github", "jira", "slack",
"vpn", "okta", "awsconsole", "gcpconsole", "databricks",
"hris", "email", "crm", "vault", "payments_api"
]
roles = ["employee", "analyst", "engineer", "manager", "admin", "contractor", "service_account"]
geos = ["IN", "US", "GB", "DE", "SG", "AE", "BR", "NL", "AU", "JP"]
privileged_users = set(rng.choice(users, size=7, replace=False))
compromisedusers = set(rng.choice(list(set(users) - privilegedusers), size=4, replace=False))
risky_devices = set(rng.choice(devices, size=5, replace=False))
risky_ips = set(rng.choice(ips, size=5, replace=False))
sensitiveservices = {"awsconsole", "gcpconsole", "vault", "paymentsapi", "snowflake"}
Usiamo questa configurazione parametrica per assegnare a ogni entità una rappresentazione logica e fisica, in modo da riflettere la realtà di un'azienda di grandi dimensioni.
Ruoli e Accessi Utente
In questo dataset, ogni utente riceve un ruolo diverso (employee, manager, admin, contractor, ecc.) nonché una geografia d'origine. La geografia e l'accesso ai servizi vengono generati in base a criteri specifici.Esempio di Codice:
user_role = {}
for u in users:
if u in privileged_users:
user_role[u] = rng.choice(["admin", "manager", "engineer"], p=[0.55, 0.2, 0.25])
elif rng.random() < 0.08:
user_role[u] = "contractor"
else:
user_role[u] = rng.choice(["employee", "analyst", "engineer"], p=[0.45, 0.25, 0.30])
userhomegeo = {u: rng.choice(geos, p=[0.30, 0.22, 0.10, 0.08, 0.08, 0.05, 0.04, 0.04, 0.04, 0.05]) for u in users}
device_owner = {d: rng.choice(users) for d in devices}
base_time = pd.Timestamp("2026-06-01 00:00:00")
events = []
for i in range(n_events):
if rng.random() < 0.18:
user = rng.choice(list(compromised_users))
else:
user = rng.choice(users)
if user in compromised_users and rng.random() < 0.42:
device = rng.choice(list(risky_devices))
else:
owned = [d for d, owner in device_owner.items() if owner == user]
device = rng.choice(owned if owned and rng.random() < 0.78 else devices)
if user in compromised_users and rng.random() < 0.50:
ip = rng.choice(list(risky_ips))
else:
ip = rng.choice(ips)
if user in compromised_users and rng.random() < 0.45:
service = rng.choice(list(sensitive_services))
else:
service = rng.choice(services)
role = user_role[user]
homegeo = userhome_geo[user]
geo = homegeo if rng.random() < 0.88 else rng.choice([g for g in geos if g != homegeo])
hour = int(rng.integers(0, 24))
minute = int(rng.integers(0, 60))
timestamp = base_time + pd.Timedelta(days=int(rng.integers(0, 10)), hours=hour, minutes=minute)
impossibletravel = int(geo != homegeo and rng.random() < 0.65)
off_hours = int(hour < 6 or hour > 21)
servicesensitivity = 1.0 if service in sensitiveservices else 0.25
privileged = int(role in ["admin", "manager", "service_account"])
compromised = int(user in compromised_users)
riskyinfra = int(device in riskydevices or ip in risky_ips)
risk_score = (
0