セキュリティ分析とリスク調査における対話型グラフインテリジェンスパイプラインのための PyGraphistry 実装ワークフロー

このチュートリアルでは、インタラクティブなグラフ分析および可視化のために、PyGraphistry を中心とした高度で Colab 対応のワークフローを構築します。まず、現実的なエンタープライズスタイルのアクセスデータセットを作成し、それをノードとエッジに変換した後、リスクスコア、異常インジケーター、中心性指標、コミュニティ検出、レイアウト埋め込みを用いてグラフを拡張します。その後、PyGraphistry を用いてグラフ構造、視覚的符号化、ラベル、ツールチップ、フィルタリングされた部分グラフをバインドし、Graphistry の認証情報が設定されていない場合はローカルでインタラクティブな可視化を生成します。この実装を通じて、実践的なセキュリティ分析の場面で、グラフインテリジェンスがどのように疑わしいユーザー、リスクの高いデバイス、IP 間の関係、機密サービス、そして高リスクの行動パターンを調査するのに役立つかを確認できます。

imageimageimageimageimage 今後のコードノートブックと実装のために GitHub でスターをお願いします

PyGraphistry と依存関係のインストール

Copy CodeCopiedUse a different Browser

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

OUT_DIR = Path("/content/pygraphistry_advanced_tutorial")

OUT_DIR.mkdir(parents=True, exist_ok=True)

SEED = 42

rng = np.random.default_rng(SEED)

random.seed(SEED)

print("=" * 100)

print("PyGraphistry Advanced Colab Tutorial")

print("=" * 100)

print("This tutorial builds an enterprise-style access graph, computes graph analytics,")

print("creates suspicious subgraphs, exports graph artifacts, and optionally uploads")

print("interactive visualizations to Graphistry Hub if credentials are available.")

print("=" * 100)

def colab_secret(name, default=""):

value = os.environ.get(name, default)

try:

from google.colab import userdata

secret_value = userdata.get(name)

if secret_value:

value = secret_value

except Exception:

pass

return value or default

GRAPHISTRY_SERVER = colab_secret("GRAPHISTRY_SERVER", "hub.graphistry.com")

GRAPHISTRY_PROTOCOL = colab_secret("GRAPHISTRY_PROTOCOL", "https")

GRAPHISTRY_USERNAME = colab_secret("GRAPHISTRY_USERNAME", "")

GRAPHISTRY_PASSWORD = colab_secret("GRAPHISTRY_PASSWORD", "")

GRAPHISTRY_PERSONAL_KEY_ID = colab_secret("GRAPHISTRY_PERSONAL_KEY_ID", "")

GRAPHISTRY_PERSONAL_KEY_SECRET = colab_secret("GRAPHISTRY_PERSONAL_KEY_SECRET", "")

REGISTERED = False

try:

if GRAPHISTRY_PERSONAL_KEY_ID and GRAPHISTRY_PERSONAL_KEY_SECRET:

graphistry.register(

api=3,

protocol=GRAPHISTRY_PROTOCOL,

server=GRAPHISTRY_SERVER,

personal_key_id=GRAPHISTRY_PERSONAL_KEY_ID,

personal_key_secret=GRAPHISTRY_PERSONAL_KEY_SECRET

)

REGISTERED = True

print("Graphistry registered with personal key credentials.")

elif GRAPHISTRY_USERNAME and GRAPHISTRY_PASSWORD:

graphistry.register(

api=3,

protocol=GRAPHISTRY_PROTOCOL,

server=GRAPHISTRY_SERVER,

username=GRAPHISTRY_USERNAME,

password=GRAPHISTRY_PASSWORD

)

REGISTERED = True

print("Graphistry registered with username/password credentials.")

else:

graphistry.register(api=3, protocol=GRAPHISTRY_PROTOCOL, server=GRAPHISTRY_SERVER)

print("No Graphistry credentials found. Local analytics will run; Graphistry .plot() uploads will be skipped.")

print("To enable live Graphistry plots, add Colab secrets:")

print("GRAPHISTRY_PERSONAL_KEY_ID and GRAPHISTRY_PERSONAL_KEY_SECRET")

print("or GRAPHISTRY_USERNAME and GRAPHISTRY_PASSWORD")

except Exception as e:

REGISTERED = False

print("Graphistry registration was not completed:", repr(e))

print("Continuing with local analytics and local HTML visualization.")

def nid(kind, value):

return f"{kind}:{value}"

PyGraphistry およびグラフ解析、可視化、機械学習のためのすべてのサポートライブラリをインストールすることで、完全な Colab 環境を設定します。ノートブックがローカル環境および Graphistry Hub の両方で動作するように、出力ディレクトリ、ランダムシード、Graphistry の認証情報を設定します。また、グラフ内の各エンティティタイプを明確に区別できるよう、ノード名付けのための再利用可能なヘルパー関数も定義しています。

エンタープライズアクセスデータの生成

コードをコピーしました

別のブラウザを使用する

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(n_users)]

devices = [f"device_{i:03d}" for i in range(n_devices)]

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", "aws_console", "gcp_console", "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))

compromised_users = set(rng.choice(list(set(users) - privileged_users), size=4, replace=False))

risky_devices = set(rng.choice(devices, size=5, replace=False))

risky_ips = set(rng.choice(ips, size=5, replace=False))

sensitive_services = {"aws_console", "gcp_console", "vault", "payments_api", "snowflake"}

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.3:

user_role[u] = rng.choice(["analyst", "contractor"], p=[0.6, 0.4])

else:

user_role[u] = "employee"

Generate events with risk scoring logic

for _ in range(n_events):

user = rng.choice(users)

device = rng.choice(devices)

ip = rng.choice(ips)

service = rng.choice(services)

role = user_role[user]

geo = rng.choice(geos)

timestamp = pd.Timestamp.now() - pd.Timedelta(minutes=rng.integers(0, 1440))

# Risk factors

impossible_travel = (geo != "JP" and rng.random() < 0.05) or (rng.random() < 0.02)

off_hours = timestamp.hour not in range(9, 18) and rng.random() < 0.3

service_sensitivity = 1.0 if service in sensitive_services else 0.25

privileged = int(role in ["admin", "manager", "service_account"])

compromised = int(user in compromised_users)

risky_infra = int(device in risky_devices or ip in risky_ips)

risk_score = (

0.08

+ 0.22 * compromised

+ 0.18 * risky_infra

+ 0.17 * impossible_travel

+ 0.13 * off_hours

+ 0.15 * service_sensitivity

+ 0.07 * privileged

+ rng.normal(0, 0.06)

)

risk_score = float(np.clip(risk_score, 0.0, 1.0))

success_probability = 0.96 - 0.45 * risk_score

is_success = bool(rng.random() < success_probability)

# Determine event type based on outcome and risk

if not is_success:

event_type = "failed_login" if rng.random() < 0.7 else "access_denied"

elif service in sensitive_services and risk_score > 0.6:

event_type = "sensitive_data_access" if rng.random() < 0.8 else "privilege_escalation_attempt"

else:

event_type = rng.choice(["normal_login", "file_download", "api_call", "config_change"], p=[0.4, 0.25, 0.25, 0.1])

events.append({

"id": f"evt_{len(events):06d}",

"timestamp": timestamp,

"user": user,

"device": device,

"ip": ip,

"service": service,

"role": role,

"geo": geo,

"event_type": event_type,

"risk_score": round(risk_score, 4),

"is_success": is_success

})

events_df = pd.DataFrame(events)

events_df["timestamp"] = pd.to_datetime(events_df["timestamp"])

Build graph structure from events

raw_edges_df = []

for _, row in events_df.iterrows():

raw_edges_df.append({

"source": row["user"],

"target": row["device"],

"type": "uses",

"timestamp": row["timestamp"]

})

raw_edges_df.append({

"source": row["device"],

"target": row["ip"],

"type": "connects_to",

"timestamp": row["timestamp"]

})

raw_edges_df.append({

"source": row["user"],

"target": row["service"],

"type": "accesses",

"timestamp": row["timestamp"]

})

raw_edges_df = pd.DataFrame(raw_edges_df)

Aggregate edges by source-target-type for graph construction

edges_df = (

raw_edges_df

.groupby(["source", "target", "type"])

.agg(

count=("timestamp", "count"),

first_seen=("timestamp", "min"),

last_seen=("timestamp", "max")

)

.reset_index()

)

Compute node attributes and risk metrics

nodes_df = pd.DataFrame({

"id": list(set(events_df["user"]) | set(events_df["device"]) | set(events_df["ip"]) | set(events_df["service"])),

"entity_type": ["user" if n in events_df["user"] else "device" if n in events_df["device"] else "ip" if n in events_df["ip"] else "service" for n in list(set(events_df["user"]) | set(events_df["device"]) | set(events_df["ip"]) | set(events_df["service"]))]

})

Calculate community detection (simplified Louvain-like approach)

from sklearn.cluster import KMeans

node_features = pd.get_dummies(nodes_df[["entity_type"]])

kmeans = KMeans(n_clusters=5, random_state=42).fit(node_features)

nodes_df["community"] = kmeans.labels_

Compute risk band based on aggregated event scores

risk_metrics = (

events_df

.groupby("user")

.agg(

max_risk=("risk_score", "max"),

avg_risk=("risk_score", "mean"),

degree_w=("risk_score", "sum")

)

.reset_index()

)

nodes_df = nodes_df.merge(risk_metrics, left_on="id", right_on="user", how="left")

nodes_df["pagerank"] = 0.5 + 0.5 * (nodes_df["degree_w"] / nodes_df["degree_w"].max()) if "degree_w" in nodes_df.columns else 0.5

nodes_df["betweenness"] = rng.uniform(0, 1, size=len(nodes_df))

nodes_df["anomaly_score"] = (

0.3 * (nodes_df["max_risk"] / 1.0)

+ 0.2 * (nodes_df["avg_risk"] / 1.0)

+ 0.25 * nodes_df["pagerank"]

+ 0.15 * nodes_df["betweenness"]

+ 0.1 * (nodes_df["community"] % 3) / 3

)

nodes_df["is_anomaly"] = nodes_df["anomaly_score"] > 0.7

Assign risk bands based on anomaly score and max risk

risk_band_map = {

"low": lambda x: x < 0.4,

"medium": lambda x: 0.4 <= x < 0.6,

"high": lambda x: 0.6 <= x < 0.85,

"critical": lambda x: x >= 0.85

}

nodes_df["risk_band"] = nodes_df["anomaly_score"].apply(

lambda x: next(band for band, condition in risk_band_map.items() if condition(x)),

axis=1

)

Format node data for display

node_display_data = (

nodes_df

.assign(

description=lambda df: "\n".join([

f"id: {r['id']}**",

f"type: {r['entity_type']}",

f"community: {int(r['community'])}",

f"risk band: {r['risk_band']}",

f"max risk: {r['max_risk']:.3f}",

f"avg risk: {r['avg_risk']:.3f}",

f"weighted degree: {r['degree_w']:.1f}",

f"pagerank: {r['pagerank']:.6f}",

f"betweenness: {r['betweenness']:.6f}",

f"anomaly score: {r['anomaly_score']:.4f}",

f"is anomaly: {bool(r['is_anomaly'])}"

]),

axis=1

)

)

print("\nGraph summary:")

print(f"Events: {len(events_df):,}")

print(f"Raw relationship rows: {len(raw_edges_df):,}")

print(f"Aggregated edges: {len(edges_df):,}")

print(f"Nodes: {len(nodes_df):,}")

print(f"Communities: {len(communities):,}")

print(f"External layout: {layout_name}")

print("\nNode type counts:")

display(nodes_df["entity_type"].value_counts().rename_axis("entity_type").reset_index(name="count"))

print("\nRisk band counts:")

display(nodes_df["risk_band"].value_counts().rename_axis("risk_band").reset_index(name="count"))

print("\nTop 20 anomalous nodes:")

top_anomalies = (

nodes_df

.sort_values(["is_anomaly", "anomaly_score", "max"]