Introduction à KQL : Kusto Query Language (2/2)

Dans cette seconde partie d’article nous allons aborder des requêtes plus complexes, avec en particulier l’usage de :

PARSE, pour parser des données en mode texte
MV-EXPAND pour développer des tableaux dynamiques
UNION et JOIN pour des requêtes portant sur plusieurs tables
L’utilisation des fonctions

PARSE et SPLIT

Certaines données de journaux collectées par Azure Monitor incluent plusieurs informations dans une seule propriété. Bien souvent le cas des Custom Logs. Il est alors nécessaire d’analyser plusieurs propriétés pour faciliter leur utilisation dans la requête

PARSE peut être utilisé avec :

Des patterns de type Texte
Des expressions régulières
Des données délimitées
Une structure prédéfinie (XML, JSON, …)

SPLIT prend une chaîne et la fractionne en sous-chaînes en fonction d’un délimiteur spécifié, retournant les sous-chaînes dans un tableau

Prenons le cas de logs envoyés vers le Workspace Log Analytics en provenance de de SYSMON. Les données peuvent être regroupées au sein d’un seul attribut et délimitées par « : » ou par une structure XML comme illustré dans le schéma ci-dessous respectivement pour les attributs RenderdDescription et EventData.

Poursuivons notre cas en souhaitant créer une Analytic Rule dans Microsoft Sentinel afin de créer une alerte s’il y a eu au moins un changement dans la base de registre au cours de la dernière semaine. Collecter des détails sur l’opération réalisée (type de modification, date, process ID, image, target objet.

Pour retenir la modification de la base de registre :

Il est nécessaire de créer une colonne contenant l’opération réalisée (dans notre cas Registry value set qui est le premier élément de la chaîne de caractère « RenderedDescription », celle-ci utilisant « : » comme séparateur

On va pour cela utiliser SPLIT comme ci-dessous, le premier élément d’une chaîne de caractère portant l’indice 0

| extend RenderedDescription = tostring(split(RenderedDescription, « : »)[0])

Pour retenir les éléments de détails la modification réalisée : Ces éléments sont dans EventData sous la forme XML. Ils sont en position 0, 1, 4, 5, 6 et 7. On va pour cela utiliser PARSE comme ci-dessous :

| extend EvData = parse_xml(EventData)
| extend EventDetail = EvData.DataItem.EventData.Data
| project-away EventData, EvData
| extend RuleName = EventDetail.[0].[« #text »], EventType = EventDetail.[1].[« #text »], ProcessId = EventDetail.[4].[« #text »], Image = EventDetail.[5].[« #text »], TargetObject = EventDetail.[6].[« #text »], Details = EventDetail.[7].[« #text »]
| project-away EventDetail

Plus de détails sur SPLIT ici : https://learn.microsoft.com/fr-fr/kusto/query/split-function?view=microsoft-fabric

Plus de détails sur PARSE ici : https://learn.microsoft.com/fr-fr/kusto/query/parse-operator?view=microsoft-fabric

MV-EXPAND

Il arrive aussi que le résultat d’une requête simple, comme des requêtes dans a table « SecurityAlert » renvoi un résultat sous la forme d’un tableau dynamique à valeurs multiples comme dans l’exemple ci-dessous :

Ainsi pour répondre à la question : « Afficher, pour chaque alerte, s’il y a une adresse IP associée et quel est l’emplacement », il est nécessaire d’isoler cette information en développant le tableau dynamique.

On peut par exemple utiliser la requête suivante :

Afin d’obtenir le résultat ci-dessous

Plus de détail sur MV-EXPAND ici : https://learn.microsoft.com/fr-fr/kusto/query/mv-expand-operator?view=microsoft-fabric

Requêtes sur de multiples tables :

Lors de l’utilisation de KQL, on en vient très rapidement à faire des requêtes sur plusieurs tables. Pour cela on utilise :

UNION : prend deux tables ou plus et renvoie les lignes de chacune d’entre elles

Exemples de requêtes :

//list of all Security Events and Security Alerts
SecurityEvent
| union SecurityAlert
|project Type, SecurityEventActivity = Activity, AlertSeverity, AlertName

//number of rows in each table whose name begins with Security
union Security*
| summarize count() by Type

JOIN : Fusionne les lignes de deux tables pour former une nouvelle table en faisant correspondre les valeurs des colonnes spécifiées de chaque table. Avec :

kind=inner : Un enregistrement de sortie est produit chaque fois qu’un enregistrement sur le côté gauche a la même clé de jointure que l’enregistrement sur le côté droit
kind=leftouter : contient toujours tous les enregistrements de la table de gauche, même si la condition de jointure ne trouve aucun enregistrement correspondant dans la table de droite.
kind=rightouter : le traitement est inversé

Exemple de syntaxe permettant de poser la question : « Au cours du mois dernier, y a-t-il des tentatives de connexion réussies et échouées sur le même ordinateur avec le même compte, quel est le nombre de chacun et dans quelle proportion »

Pour cela il est nécessaire de faire une jointure sur la même table afin de récupérer les deux évènements 4624 et 4625 :

SecurityEvent
| where EventID == « 4624 » and TimeGenerated > ago(30d)
| summarize SuccessLogOnCount=count() by EventID, Account, Computer
| project SuccessLogOnCount, Account , Computer
| join kind = inner (
     SecurityEvent
     | where EventID == « 4625 » and TimeGenerated > ago(30d)
     | summarize FailedLogOnCount=count() by EventID, Account, Computer
     | project FailedLogOnCount, Account , Computer
) on Account
|project-away Account1, Computer1
|extend proportion = FailedLogOnCount / SuccessLogOnCount
|order by proportion desc

Utilisation des fonctions :

Il existe deux types de fonction.

Les fonctions stockées (utilisables par toutes requêtes). Ce sont des fonctions définies par l’utilisateur qui sont stockées et gérées dans les entités de schéma de la base de données, utilization de “Save as function”.

Les fonctions définies dans la requête elle-même. Ce sont des fonctions définies par l’utilisateur qui sont définies et utilisées dans le cadre d’une requête à l’aide de l’instruction let.

Exemple de syntaxe pour « Déterminer s’il y a une attaque par dictionnaire sur les ordinateurs avec les mises à jour de sécurité nécessaires »

let machinesWithUpdateNeeded = (){
    Update
        |where TimeGenerated >= ago(30d)
        |where UpdateState has « Needed » and Title contains « security »
        |summarize Computer=makeset(Computer)
};
SecurityDetection
|where TimeGenerated >= ago(100d)
| where Description contains « dictionary attack »
| where Computer in (machinesWithUpdateNeeded) //limit the query to just these computers
| summarize by Computer

Requête portant sur plusieurs workspace :

KQL prend en charge l’interrogation sur plusieurs espaces de travail Log Analytics, sur le même groupe de ressources, un autre groupe de ressources ou un autre abonnement

Il faut pour cela préfixer la table par le nom du workspace :

let unionSecurityEvent = () {
union (workspace(« azsecusentinel »).SecurityAlert), (workspace(« sc200DemoWKS »).SecurityAlert)
};
unionSecurityEvent
|where TimeGenerated > ago(30d)
|summarize count() by AlertSeverity, bin(TimeGenerated, 1d)
|render timechart

Optimisation des requêtes :

Comme le montre les copies ci-dessous, il est important de bien concevoir ses requêtes afin d’optimiser le temps de traitement. En particulier :