Dans cette série de blogs en quatre parties « Leçons tirées de la construction de Lakehouses de cybersécurité » nous discuterons d’un certain nombre de défis auxquels les organisations sont confrontées en matière d’ingénierie des données lors de la création d’un Lakehouse pour les données de cybersécurité, et proposerons des solutions, des conseils, des astuces et des bonnes pratiques que nous avons utilisés sur le terrain pour les surmonter. Si vous souhaitez construire votre propre Lakehouse de cybersécurité, cette série vous informera sur les défis et vous proposera une voie à suivre.
Databricks a créé une solution de configuration pratique à faible code pour gérer et standardiser efficacement les cyber-journaux. Notre plateforme Lakehouse simplifie l’ingénierie des données, facilitant une transition plus rapide vers la recherche, l’analyse et la détection des menaces en continu. Il complète vos systèmes SIEM et SOAR existants, améliorant ainsi vos opérations de cybersécurité sans complexité inutile.
Dans la première partie, nous commençons par l’élément le plus fondamental de tout moteur de cyberanalyse : extraction uniforme de l’horodatage des événements. Des horodatages précis font partie des éléments les plus importants dans les opérations de sécurité et la réponse aux incidents. Sans précision, il est impossible de générer une séquence d’événements pris par les utilisateurs du système ou par de mauvais acteurs. Dans ce blog, nous examinerons certaines des stratégies disponibles pour identifier, extraire et transformer les informations d’horodatage des événements en Delta Lake, de manière à ce qu’elles soient utilisables dans un contexte cybernétique.
Pourquoi l’heure de l’événement est-elle si importante ?
Les données de journal générées par la machine sont, au mieux, compliquées. Il existe des structures bien définies pour des types de fichiers spécifiques (JSON, YAML, CSV, etc.), mais le contenu et le format des données qui composent ces fichiers sont en grande partie laissés à l’interprétation des développeurs. Bien que des formats d’heure existent (OIN 8601), leur respect est limité et subjectif – peut-être que les formats de journaux sont antérieurs à ces normes, ou que la préférence géographique pour un format spécifique détermine la façon dont ces horodatages sont écrits.
Malgré les formats horaires variés signalés dans les journaux, nous sommes responsables de leur normalisation pour garantir l’interopérabilité avec toutes les données de journaux reçues et analysées dans n’importe quel cyber-moteur.
Pour souligner l’importance de l’interopérabilité entre les horodatages, considérons certaines des tâches auxquelles un centre d’opérations de sécurité (SOC) typique doit répondre quotidiennement.
- Quel ordinateur l’attaquant a-t-il compromis en premier ?
- Dans quel ordre l’attaquant s’est-il déplacé d’un système à l’autre ?
- Quelles activités se sont déroulées et dans quel ordre une fois le point d’ancrage initial établi ?
Sans horodatages précis et unifiés, il est impossible de générer une chronologie des activités survenues pour répondre efficacement à ces questions. Ci-dessous, nous examinons certains des défis et proposons des conseils sur la manière de les aborder.
Problèmes d’horodatage
Colonne multiple ou unique : Avant d’envisager comment analyser un horodatage d’événement, nous devons d’abord l’isoler. Cela peut déjà se produire automatiquement dans certains formats de journaux ou dans des opérations de lecture instantanée. Cependant, dans d’autres cas, c’est peu probable. Par exemple, les fichiers CSV (valeurs séparées par des virgules) seront extraits par Spark sous forme de colonnes individuelles. Si l’horodatage est isolé par l’un d’entre eux, alors tant mieux ! Cependant, une machine générant des données Syslog apparaîtra probablement sous la forme d’une seule colonne et l’horodatage doit être isolé à l’aide d’expressions régulières.
Formats de date et d’heure : Cela provoque beaucoup de confusion dans les fichiers journaux. Par exemple, « 06/12/12 » contre « 12/06/12 ». Les deux formats sont valides, mais il est difficile d’identifier le jour, le mois et l’année sans connaître le format du journal système local.
Identification du fuseau horaire : Semblable aux formats de données et d’heure, certains systèmes indiquent le fuseau horaire de l’horodatage, tandis que d’autres supposent une heure locale et n’impriment pas du tout le fuseau horaire. Cela ne pose peut-être pas de problème si toutes les sources de données sont signalées et analysées dans le même fuseau horaire. Cependant, les organisations doivent analyser des dizaines, voire des centaines de sources de journaux provenant de plusieurs fuseaux horaires dans le monde connecté et global d’aujourd’hui.
L’identification, l’extraction et l’analyse des horodatages des événements nécessitent une représentation cohérente et efficace du temps au sein de nos systèmes de stockage. Vous trouverez ci-dessous un exemple de comment extraire et analyser un horodatage à partir d’un serveur Web Apache de style syslog.
Scénario d’extraction d’horodatages
Dans l’exemple suivant, nous examinons le format de journal standard du serveur Web Apache. Les données sont générées sous forme d’enregistrement de texte et sont lues comme une seule colonne (valeur) dans Databricks. Par conséquent, nous devons extraire l’horodatage de l’événement à l’aide d’une expression régulière.
Exemple d’expression régulière pour extraire l’horodatage de l’événement à partir d’une seule colonne de données :
from pyspark.sql.functions import regexp_extract
TIMESTAMP_REGEX = '^([^ ]*) [^ ]* ([^ ]*) \[([^\]]*)\]'
df1 = df.select(regexp_extract("value", TIMESTAMP_REGEX, 3).alias('_raw_time'), "*")
display(df1)Nous utilisons le PySpark regexp_extract fonction pour extraire la partie de la chaîne qui a l’horodatage de l’événement et créer une colonne _raw_time avec les personnages correspondants.
Trame de données résultante :
Analyse des horodatages
Avec l’horodatage de l’événement extrait sous forme de nouvelle colonne, nous pouvons désormais le normaliser en un horodatage standard ISO 8601.
Pour normaliser l’horodatage, nous devons définir le format à l’aide du format date/heure modificateurs et convertissez-le en un horodatage de style Unix avant de le transformer au format d’horodatage au format ISO.
TIMESTAMP_FORMAT = "dd/MMM/yyyy:HH:mm:ss Z"Exemple de transformation vers un horodatage d’événement au format ISO 8601 :
from pyspark.sql.functions import to_timestamp, unix_timestamp, col
TIMESTAMP_FORMAT="dd/MMM/yyyy:HH:mm:ss Z"
df2 = df1.select(
to_timestamp(unix_timestamp(col("_raw_time"), TIMESTAMP_FORMAT).cast("timestamp"), "dd-MM-yyyy HH:mm:ss.SSSZ").alias("_event_time")
)
display(df2)Nous utilisons les fonctions PySpark unix_timestamp et to_timestamp pour générer la nouvelle colonne de métadonnées _event_time.
Trame de données résultante :
La colonne résultante est convertie en type d’horodatage pour garantir la cohérence et l’intégrité des données.
Conseils et bonnes pratiques
Dans le cadre de notre démarche visant à aider de nombreux clients en matière de cyberanalyse, nous avons rassemblé des conseils précieux et des bonnes pratiques qui peuvent améliorer considérablement l’expérience d’ingestion.
Format d’heure explicite : Lors de la création d’analyseurs, la définition explicite du format d’heure accélérera considérablement la tâche d’analyse par rapport au passage d’une colonne à une bibliothèque générique qui doit tester de nombreux formats pour en trouver un qui renvoie une colonne d’horodatage précise.
Nom des colonnes : Préfixez les colonnes de métadonnées avec un trait de soulignement. Cela permet une distinction facile entre les données générées par la machine et les métadonnées, avec l’avantage supplémentaire d’apparaître justifié à gauche par défaut dans les blocs de données et les tableaux.
Heure de l’événement par rapport à l’heure d’ingestion : Des retards se produisent dans la transmission des données. Ajoutez une nouvelle colonne de métadonnées pour le temps d’ingestion et créez une rigueur opérationnelle pour identifier les sources de données actuellement en retard ou manquantes.
Valeurs par défaut : Élaborez une stratégie sur les horodatages manquants ou indéterminés. Les choses peuvent mal tourner et vont mal. Faites un jugement sur la façon de traiter les horodatages manquants. Certaines des tactiques que nous avons vues sont :
- Remettre la date à zéro (01/01/1970) et créer une rigueur opérationnelle pour identifier et corriger les données.
- Réglez la date sur l’heure d’ingestion actuelle et créez une rigueur opérationnelle pour identifier et corriger les données
- Échouer complètement le pipeline
Conclusion
Des horodatages d’événements bien formés et précis sont essentiels aux opérations de sécurité de l’entreprise et à la réponse aux incidents afin de générer des séquences d’événements et des calendriers pour enquêter sur les cybermenaces. Sans interopérabilité entre toutes les sources de données, il est impossible de maintenir une posture de sécurité efficace. Des complexités telles que l’extraction d’expressions régulières et l’analyse des divergences dans les sources de données sous-tendent cela. En aidant de nombreux clients à créer des Lakehouses de cybersécurité, nous avons créé des solutions pratiques pour accélérer ce processus.
Entrer en contact
Dans ce blog, nous avons examiné un seul exemple des nombreux problèmes d’extraction d’horodatage possibles rencontrés avec les fichiers journaux semi-structurés. Si vous souhaitez en savoir plus sur la façon dont les cybersolutions Databricks peuvent permettre à votre organisation d’identifier et d’atténuer les cybermenaces, contactez [email protected] et découvrez notre nouveau Page Web de Lakehouse pour les applications de cybersécurité.
