graph_blast_radius_fl()

2025-05-25

適用対象: ✅Microsoft Fabric✅Azure Data Explorer✅Azure Monitor✅Microsoft Sentinel

パスまたはエッジデータ上のソースノードのブラスト半径 (リストとスコア) を計算します。

関数 graph_blast_radius_fl() は、パスまたはエッジデータに基づいて各ソースノードのブラスト半径を計算できる UDF (ユーザー定義関数) です。入力データの各行には、ソースノードとターゲットノードが含まれています。これは、ノードとターゲット間の直接接続 (エッジ) またはそれらの間のより長いマルチホップパスを表すことができます。パスが使用できない場合は、まず、graph-match 演算子または graph_path_discovery_fl() 関数を使用してそれらを検出できます。その後、パス検出の出力の上に graph_blast_radius_fl() を実行できます。

Blast Radius は、関連するターゲットへの特定のソースノードの接続を表します。ソースがアクセスできるターゲットが多いほど、攻撃者によって侵害された場合の影響が大きいため、名前になります。 Blast Radius が高いノードは、サイバーセキュリティドメインにおいて重要です。これは、攻撃者が引き起こす可能性のある損害や、高い評価を受けている可能性があるためです。したがって、アラートなどのセキュリティ信号のセキュリティ強化と優先順位付けの観点から、Blast Radius が高いノードは、それに応じて保護する必要があります。

この関数は、各ソースの接続されたターゲットの一覧と、ターゲットの数を表すスコアを出力します。必要に応じて、ターゲットごとに意味のある "重み" がある場合 (重要度やコストなど)、重み付けスコアはターゲットの重みの合計として計算されます。さらに、表示されるソースの最大数と各リスト内のターゲットの最大数の制限は、より適切な制御のために省略可能なパラメーターとして公開されます。

構文

graph_blast_radius_fl( sourceIdColumnName, targetIdColumnName, [targetWeightColumnName], [resultCountLimit], [listedIdsLimit] ])

構文規則の詳細について説明します。

パラメーター

名前	種類	必須	形容
sourceIdColumnName をする	`string`	✔️	ソースノード ID を含む列の名前 (エッジまたはパスの場合)。
targetIdColumnName をする	`string`	✔️	ターゲットノード ID を含む列の名前 (エッジまたはパスの場合)。
targetWeightColumnName をする	`string`		ターゲットノードの重みを含む列の名前 (重要度など)。関連する重みが存在しない場合、加重スコアは 0 になります。既定の列名は noWeightsColumnです。
resultCountLimit をする	`long`		返される行の最大数 (降順スコアで並べ替えられます)。既定値は 100000 です。
listedIdsLimit をする	`long`		各ソースに対して一覧表示されるターゲットの最大数。既定値は 50 です。

関数の定義

関数を定義するには、次のようにコードをクエリ定義関数として埋め込むか、データベースに格納された関数として作成します。

クエリ定義の
保存されている

次の let ステートメント使用して関数を定義します。アクセス許可は必要ありません。

大事な

let ステートメント単独では実行できません。その後に、表形式の式ステートメント続ける必要があります。 graph_blast_radius_fl()の作業例を実行するには、例を参照してください。

let graph_blast_radius_fl = (T:(*), sourceIdColumnName:string, targetIdColumnName:string, targetWeightColumnName:string = 'noWeightsColumn'
    , resultCountLimit:long = 100000, listedIdsLimit:long = 50)
{
let paths = (
    T
    | extend sourceId           = column_ifexists(sourceIdColumnName, '')
    | extend targetId           = column_ifexists(targetIdColumnName, '')
    | extend targetWeight       = tolong(column_ifexists(targetWeightColumnName, 0))
);
let aggregatedPaths = (
    paths
    | sort by sourceId, targetWeight desc
    | summarize blastRadiusList = array_slice(make_set_if(targetId, isnotempty(targetId)), 0, (listedIdsLimit - 1))
                , blastRadiusScore = dcountif(targetId, isnotempty(targetId))
                , blastRadiusScoreWeighted = sum(targetWeight)
        by sourceId
    | extend isBlastRadiusListCapped = (blastRadiusScore > listedIdsLimit)
);
aggregatedPaths
| top resultCountLimit by blastRadiusScore desc
};
// Write your query to use the function here.

次の .create functionを使用して、ストアド関数を 1 回定義します。データベースユーザーのアクセス許可が必要です。

大事な

例のに示すように、関数を使用する前に、このコードを実行して関数を作成する必要があります。

.create-or-alter function with (docstring = "Calculate the Blast Radius (list and score) of source nodes over path or edge data", skipvalidation = "true", folder = 'Cybersecurity') 
graph_blast_radius_fl (T:(*), sourceIdColumnName:string, targetIdColumnName:string, targetWeightColumnName:string = 'noWeightsColumn'
    , resultCountLimit:long = 100000, listedIdsLimit:long = 50)
{
let paths = (
    T
    | extend sourceId           = column_ifexists(sourceIdColumnName, '')
    | extend targetId           = column_ifexists(targetIdColumnName, '')
    | extend targetWeight       = tolong(column_ifexists(targetWeightColumnName, 0))
);
let aggregatedPaths = (
    paths
    | sort by sourceId, targetWeight desc
    | summarize blastRadiusList = array_slice(make_set_if(targetId, isnotempty(targetId)), 0, (listedIdsLimit - 1))
                , blastRadiusScore = dcountif(targetId, isnotempty(targetId))
                , blastRadiusScoreWeighted = sum(targetWeight)
        by sourceId
    | extend isBlastRadiusListCapped = (blastRadiusScore > listedIdsLimit)
);
aggregatedPaths
| top resultCountLimit by blastRadiusScore desc
}

例

次の例では、呼び出し演算子を使用して関数を実行します。

クエリ定義の
保存されている

クエリ定義関数を使用するには、埋め込み関数定義の後に呼び出します。

クエリを実行する

let connections = datatable (SourceNodeName:string, TargetNodeName:string, TargetNodeCriticality:int)[						
    'vm-work-1',            'webapp-prd', 	          3,
    'vm-custom',        	'webapp-prd', 	          3,
    'webapp-prd',           'vm-custom', 	          1,
    'webapp-prd',       	'test-machine', 	      1,
    'vm-custom',        	'server-0126', 	          1,
    'vm-custom',        	'hub_router', 	          2,
    'webapp-prd',       	'hub_router', 	          2,
    'test-machine',       	'vm-custom',              1,
    'test-machine',        	'hub_router', 	          2,
    'hub_router',           'remote_DT', 	          1,
    'vm-work-1',            'storage_main_backup', 	  5,
    'hub_router',           'vm-work-2', 	          1,
    'vm-work-2',        	'backup_prc', 	          3,
    'remote_DT',            'backup_prc', 	          3,
    'backup_prc',           'storage_main_backup', 	  5,
    'backup_prc',           'storage_DevBox', 	      1,
    'device_A1',            'sevice_B2', 	          2,
    'sevice_B2',            'device_A1', 	          2
];
let graph_blast_radius_fl = (T:(*), sourceIdColumnName:string, targetIdColumnName:string, targetWeightColumnName:string = 'noWeightsColumn'
    , resultCountLimit:long = 100000, listedIdsLimit:long = 50)
{
let paths = (
    T
    | extend sourceId           = column_ifexists(sourceIdColumnName, '')
    | extend targetId           = column_ifexists(targetIdColumnName, '')
    | extend targetWeight       = tolong(column_ifexists(targetWeightColumnName, 0))
);
let aggregatedPaths = (
    paths
    | sort by sourceId, targetWeight desc
    | summarize blastRadiusList = array_slice(make_set_if(targetId, isnotempty(targetId)), 0, (listedIdsLimit - 1))
                , blastRadiusScore = dcountif(targetId, isnotempty(targetId))
                , blastRadiusScoreWeighted = sum(targetWeight)
        by sourceId
    | extend isBlastRadiusListCapped = (blastRadiusScore > listedIdsLimit)
);
aggregatedPaths
| top resultCountLimit by blastRadiusScore desc
};
connections
| invoke graph_blast_radius_fl(sourceIdColumnName 		= 'SourceNodeName'
                            , targetIdColumnName 		= 'TargetNodeName'
                            , targetWeightColumnName 	= 'TargetNodeCriticality'
)

大事な

この例を正常に実行するには、最初に関数定義コードを実行して関数を格納する必要があります。

let connections = datatable (SourceNodeName:string, TargetNodeName:string, TargetNodeCriticality:int)[
    'vm-work-1',            'webapp-prd',           3,
    'vm-custom',            'webapp-prd',           3,
    'webapp-prd',           'vm-custom',            1,
    'webapp-prd',          'test-machine',          1,
    'vm-custom',           'server-0126',           1,
    'vm-custom',           'hub_router',            2,
    'webapp-prd',          'hub_router',            2,
    'test-machine',        'vm-custom',             1,
    'test-machine',        'hub_router',            2,
    'hub_router',           'remote_DT',            1,
    'vm-work-1',            'storage_main_backup',  5,
    'hub_router',           'vm-work-2',            1,
    'vm-work-2',            'backup_prc',           3,
    'remote_DT',            'backup_prc',           3,
    'backup_prc',           'storage_main_backup',  5,
    'backup_prc',           'storage_DevBox',       1,
    'device_A1',            'sevice_B2',            2,
    'sevice_B2',            'device_A1',            2
];
connections
| invoke graph_blast_radius_fl(sourceIdColumnName       = 'SourceNodeName'
                            , targetIdColumnName        = 'TargetNodeName'
                            , targetWeightColumnName    = 'TargetNodeCriticality'
)

出力

ソースID	blastRadiusList	blastRadiusScore	blastRadiusScoreWeighted	isBlastRadiusListCapped
webapp-prd	["vm-custom","test-machine","hub_router"]	3	4	偽
vm-custom	["webapp-prd","server-0126","hub_router"]	3	6	偽
test-machine	["vm-custom","hub_router"]	2	3	偽
vm-work-1	["webapp-prd","storage_main_backup"]	2	8	偽
backup_prc	["storage_main_backup","storage_DevBox"]	2	6	偽
hub_router	["remote_DT","vm-work-2"]	2	2	偽
vm-work-2	["backup_prc"]	1	3	偽
device_A1	["sevice_B2"]	1	2	偽
remote_DT	["backup_prc"]	1	3	偽
sevice_B2	["device_A1"]	1	2	偽

この関数を実行すると、ソースとターゲットの間の接続またはパスがソースごとに集計されます。各ソースについて、Blast Radius は接続されたターゲットをスコア (標準および重み付け) およびリストとして表します。

出力の各行には、次のフィールドが含まれています。

sourceId: 関連する列から取得されたソースノードの ID。
blastRadiusList: ソースノードが接続されているターゲットノード ID (関連する列から取得) の一覧。リストは、listedIdsLimit パラメーターの最大長制限に制限されます。
blastRadiusScore: スコアは、ソースが接続されているターゲットノードの数です。高いブラスト半径スコアは、ソースノードが多数のターゲットにアクセスできる可能性があることを示し、それに応じて処理する必要があります。
blastRadiusScoreWeighted: 重み付けスコアは、オプションのターゲットノードの重み列の合計であり、その値 (重要度やコストなど) を表します。このような重みが存在する場合、高い値のターゲットにアクセスする可能性があるため、加重された Blast Radius スコアがソースノード値のより正確なメトリックになる可能性があります。
isBlastRadiusListCapped: ターゲットのリストが listedIdsLimit パラメーターによって上限に達したかどうかを示すブール値フラグ。 true の場合は、リストされているターゲット (blastRadiusScore の数まで) に加えて、ソースから他のターゲットにアクセスできます。

上記の例では、ソースとターゲット間の接続に対して graph_blast_radius_fl() 関数を実行します。出力の最初の行では、ソースノード 'webapp-prd' が 3 つのターゲット ('vm-custom'、'test-machine'、'hub_router') に接続されていることがわかります。ターゲットの重みとして入力データ TargetNodeCriticality 列を使用し、累積重み 4 を取得します。また、ターゲットの数は 3 で、既定のリスト制限は 50 であるため、すべてのターゲットが表示されるため、isBlastRadiusListCapped 列の値は FALSE になります。

マルチホップパスが使用できない場合は、ソースとターゲットの間にマルチホップパスを作成し (たとえば、'graph_path_discovery_fl()') を実行し、結果の上に 'graph_blast_radius_fl()' を実行できます。

出力は似ていますが、マルチホップパスで計算された Blast Radius を表しているため、関連するターゲットへのソースノードの真の接続性をより適切に示します。ソースとターゲットのシナリオ間の完全なパス (中断など) を見つけるには、graph_path_discovery_fl() 関数を関連するソースノードとターゲットノードのフィルターと共に使用できます。

graph_blast_radius_fl() 関数を使用して、直接エッジまたは長いパスで計算されたソースノードのブラスト半径を計算できます。サイバーセキュリティドメインでは、いくつかの分析情報を提供できます。通常および重み付けされた Blast Radius スコアは、防御側と攻撃者の両方の観点からソースノードの重要性を表します。高いブラスト半径を持つノードは、アクセスの強化や脆弱性管理などの観点から、それに応じて保護する必要があります。このようなノード上のアラートなどのセキュリティシグナルに優先順位を付ける必要があります。 Blast Radius リストは、ソースとターゲット間の望ましくない接続を監視し、中断シナリオで使用する必要があります。たとえば、ソースが侵害された場合は、ソースと重要なターゲットの間の接続を切断する必要があります。

次の方法で共有

graph_blast_radius_fl()

構文

パラメーター

関数の定義

例

関連コンテンツ

フィードバック

その他のリソース