أتمتة تنبيه ثغرات Python عبر OSV وTelegram
هتخرج من المقال بأوتوميشن صغير يكشف الحزم الضعيفة في مشروع Python قبل ما تتحول لتذكرة طوارئ.
مستوى القارئ: متوسط
المشكلة باختصار
اللي بيحصل فعلاً إن فريق صغير يراجع تحديثات الحزم وقت ما يحصل عطل أو تحذير أمني كبير. الطريقة دي بتفشل لأن الثغرات لا تنتظر sprint التخطيط. لو عندك API داخلي عليه 50K طلب يوميًا، حزمة واحدة قديمة في طبقة HTTP أو parsing ممكن تفتح مخاطرة أكبر من حجم الكود اللي كتبته بنفسك.
الهدف هنا مش بناء منصة Security كاملة. الهدف أوتوميشن يومي يقرأ ملف requirements.txt، يسأل OSV.dev عن الإصدارات المثبتة، ثم يرسل تنبيه Telegram مختصر لو فيه نتيجة مهمة.
الفكرة بمثال واضح
ركز في المثال ده. عندك ملف حزم فيه:
requests==2.19.0
urllib3==1.24.1
fastapi==0.110.0بدل ما تفتح صفحات advisories يدويًا كل أسبوع، السكربت يحوّل كل سطر إلى سؤال: هل هذه الحزمة بهذا الإصدار عندها ثغرات معروفة في قاعدة OSV؟ لو الإجابة نعم، يبعت رسالة فيها اسم الحزمة وعدد النتائج وأول معرف للثغرة.
المفهوم العلمي هنا بسيط: OSV.dev يقدم API موحدًا للثغرات مفتوحة المصدر. endpoint مثل /v1/querybatch يسمح بإرسال أكثر من package/version في طلب واحد، وده أفضل من 80 طلب منفصل لو مشروعك متوسط الحجم.
السكربت العملي
الافتراض إن مشروعك Python، والملف الأساسي هو requirements.txt، والتنبيه يروح Telegram. لا تضع token في الكود. استخدم environment variables.
import os
import re
import requests
REQ_FILE = "requirements.txt"
OSV_URL = "https://api.osv.dev/v1/querybatch"
TG_TOKEN = os.environ["TELEGRAM_BOT_TOKEN"]
TG_CHAT_ID = os.environ["TELEGRAM_CHAT_ID"]
def parse_requirements(path):
items = []
pattern = re.compile(r"^([A-Za-z0-9_.-]+)==([^\s#]+)")
with open(path, "r", encoding="utf-8") as f:
for line in f:
match = pattern.match(line.strip())
if match:
name, version = match.groups()
items.append({
"version": version,
"package": {"name": name, "ecosystem": "PyPI"}
})
return items
def query_osv(queries):
response = requests.post(OSV_URL, json={"queries": queries}, timeout=30)
response.raise_for_status()
return response.json().get("results", [])
def send_telegram(message):
url = f"https://api.telegram.org/bot{TG_TOKEN}/sendMessage"
response = requests.post(url, json={
"chat_id": TG_CHAT_ID,
"text": message,
"disable_web_page_preview": True
}, timeout=20)
response.raise_for_status()
def main():
queries = parse_requirements(REQ_FILE)
if not queries:
return
results = query_osv(queries)
lines = []
for item, result in zip(queries, results):
vulns = result.get("vulns", [])
if not vulns:
continue
package = item["package"]["name"]
version = item["version"]
first_id = vulns[0].get("id", "unknown")
lines.append(f"{package}=={version}: {len(vulns)} findings, first: {first_id}")
if lines:
send_telegram("Python dependency alerts:\n" + "\n".join(lines[:20]))
if __name__ == "__main__":
main()