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

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

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

背景や根拠まで確認しますか？

調べる

選ぶ

サイト