الطريقة المفرطة في التعقيد التي يعمل بها CloudTrail و CloudWatch Events معًا

من أكثر المسائل إرباكًا في رحلتي السحابية كان فهم كيفية عمل CloudTrail و CloudWatch Events معًا. لسبب ما استغرق مني الأمر سنوات (وقدرًا كبيرًا من الاختبارات) لاستيعاب آلية عمل هذا الارتباط فعليًا؛ وبخاصة كيفية عمله مع مفهوم المناطق المتعددة ومسارات AWS Organization. وبعد أن توصلت إلى ذلك كله، افترضت أن الجميع يعرفه مسبقًا، غير أن نقاشات حديثة أوضحت أن هذا الالتباس شائع إلى حد بعيد. وفي ما يلي أفضل محاولة لديّ لتبسيط الأمور.

أولًا، المشكلة التي نحاول حلها: إنشاء قواعد CloudWatch استنادًا إلى أحداث CloudTrail واستخدامها لإرسال الإشعارات أو تشغيل دوال Lambda.

هذا كل شيء — أريد إرسال إشعار بشأن أمر بسيط مثل استدعاء واجهة برمجة التطبيقات لفتح قاعدة جديدة في مجموعة أمان (AuthorizeSecurityGroupIngress، إن كنتم تتساءلون).

ولتحقيق ذلك، تحتاجون إلى توافر ما يلي:

• تفعيل CloudTrail في المنطقة التي يجري فيها استدعاء واجهة برمجة التطبيقات.
• بث CloudTrail إلى CloudWatch.
• قاعدة CloudWatch في منطقة استدعاء واجهة برمجة التطبيقات تبحث عن ذلك الاستدعاء تحديدًا، أو عن جميع استدعاءات واجهة برمجة التطبيقات في CloudTrail.

والآن مصدر الالتباس:

• عند إنشاء مسار CloudTrail متعدد المناطق أو مسار Organization، تقوم AWS خلف الكواليس فعليًا بإعداد مسارات في كل منطقة على حدة (وفي كل حساب، في حالة مسار Organization). وهي جميعها مسارات منفصلة، لكن كل واحد منها مُهيأ لإرسال نتائجه إلى حاوية S3 مشتركة، ولا يمكن إدارة كل منها إلا في حسابه ومنطقته الأصليين.
• غير أن أحداث CloudWatch الخاصة باستدعاءات واجهة برمجة التطبيقات لا تُنشأ إلا في منطقة استدعاء واجهة برمجة التطبيقات.
• لذا إذا أنشأتم مسارًا متعدد المناطق، تُجمع البيانات كلها مركزيًا، لكن الأحداث تظهر محليًا فقط. وقاعدة CloudWatch في منطقة المسار الأصلي لن تُشغَّل إلا لاستدعاءات واجهة برمجة التطبيقات التي تجري في تلك المنطقة (الأصلية). لذلك إذا أنشأتم تنبيهًا لتغييرات مجموعات الأمان، فلن يعمل إلا في المنطقة الأصلية — لا في المناطق الأخرى — حتى مع تفعيل CloudTrail.
• ستظهر مجموعة/دفق سجلات CloudWatch في المنطقة الرئيسية، لا في المناطق الأخرى، لكن كل حدث يُنشأ في المنطقة التي أطلقته.
• وإذا أردتم جمع كل الأحداث الخاصة باستدعاءات واجهة برمجة التطبيقات، فعليكم استخدام تعريف حدث غير موثّق (أدرجته أدناه).
• إذا قرأتم وثائق Amazon… فهي لا توضّح أيًا من هذا بجلاء. أو على الأقل لم أتمكن من العثور على ذلك. بل إنني كنت ذات مرة في مكالمة دعم توصلت خلالها إلى الأمر، وظل ممثل AWS يتمتم: «لا أظن أنها تعمل بهذه الطريقة» بينما بدأت أحداثي تتدفق. كانت ليلة عصيبة بالنسبة إليه.

كان هذا غير بديهي بالنسبة إليّ لسبب ما. فقد افترضت أنكم إذا مركزتم المسار، يمكنكم عندئذ مركزة قاعدة CloudWatch لتُشغَّل بناءً على أحداث استدعاءات واجهة برمجة التطبيقات. لكن ذلك كان خاطئًا تمامًا لسوء الحظ، وحتى عند مركزة المسار، لا يزال عليكم إنشاء القاعدة في كل منطقة تهمكم. وحتى إذا استخدمتم Event Bus لجمع الأحداث من حسابات متعددة، فإنكم لا تزالون بحاجة إلى إنشاء قاعدة CloudWatch في كل منطقة من كل حساب لإرسال الحدث إلى Bus، ثم بحاجة إلى إنشاء قواعد في كل منطقة من مناطق Event Bus لتشغيل الإشعار/الإجراء الذي تريدونه.

وفي ما يلي الأسلوب الذي أوصي به إذا كنتم تريدون قدرات التنبيه شبه الفوري أو المعالجة/الإجراءات التلقائية التي تدعمها قواعد CloudWatch:

• فعّلوا مسارًا متعدد المناطق. يكفي القيام بذلك مرة واحدة، ومسار Organization كافٍ.
• يؤدي هذا إلى إنشاء جميع المسارات الإقليمية التي تحتاجون إليها. يبدو الأمر وكأنه مسار مركزي واحد، لكنه في الحقيقة مجموعة من المسارات الإقليمية ترسل بياناتها إلى متلقٍ مركزي.
• الخيار 1: أنشئوا قاعدة CloudWatch في كل منطقة تريدون فيها تنبيهًا شبه فوري. وهنا يكون CloudFormation و Terraform خير معين.
• الخيار 2: مركزوا جميع أحداثكم. أنشئوا داخل كل منطقة قاعدة لإرسال جميع أحداث CloudTrail إلى دالة Lambda أو موضوع SNS، يتولى بعد ذلك تمريرها إلى وجهتكم. ونحن نستخدم هذا الأسلوب بأنفسنا؛ إذ نرسل عبر نقطة نهاية مخصصة لواجهة برمجة التطبيقات، لكن يمكنكم البث إلى Kinesis أو إلى أي وجهة تقريبًا.

ولبدء مسيرتكم، إليكم نموذجين برمجيين.

نمط التصفية السري لقاعدة CloudWatch لديكم لجمع جميع الأحداث من CloudTrail:

{

"detail-type": [

"AWS API Call via CloudTrail"

] }

وفي ما يلي نموذج لكود Lambda لتمرير الأحداث، إلى Kinesis في هذه الحالة:

import json

import boto3

def lambda_handler(event, context):

kinesis = boto3.client('kinesis', region_name='us-west-2')

data = json.dumps(event)

print(data)

response = kinesis.put_record(

StreamName='cloudsec_prod_stream_alert_kinesis',

Data=data,

PartitionKey='test-client-id'

)

print(response)

return {

'statusCode': 200,

'body': json.dumps('Record added')

}