Loading
Loading
Secrets Management is managing keys and secrets securely — one of the most common vulnerabilities in AI projects.
Putting Secrets directly in code or in Git! Thousands of API keys are leaked daily on GitHub.
Secrets must be changed periodically — at least every 90 days, and immediately when leakage is suspected.
Use git-secrets or truffleHog to detect leaked Secrets in Git.
import re
import os
import hashlib
import base64
from dataclasses import dataclass, field
from typing import Dict, List, Optional
from datetime import datetime, timedelta
# ─── Secret Patterns (للكشف عن التسريب) ──────────────────
SECRET_PATTERNS = {
"Claude API Key": r"sk-ant-[a-zA-Z0-9-]{20,}",
"OpenAI API Key": r"sk-[a-zA-Z0-9]{48}",
"AWS Access Key": r"AKIA[A-Z0-9]{16}",
"AWS Secret": r"[a-zA-Z0-9/+=]{40}",
"Generic Password": r"passwords*=s*['"][^'"]{6,}['"]",
"Generic API Key": r"api.?keys*=s*['"][^'"]{10,}['"]",
"DB Connection": r"(mysql|postgresql|mongodb)://w+:w+@",
}
class SecretScanner:
def scan_code(self, code: str, filename: str = "unknown") -> List[Dict]:
findings = []
for name, pattern in SECRET_PATTERNS.items():
for match in re.finditer(pattern, code, re.IGNORECASE):
findings.append({
"file": filename,
"type": name,
"snippet": match.group()[:30] + "...",
"pos": match.start(),
})
return findings
# ─── Vault (Secrets Manager Simulation) ────────────────────
@dataclass
class Secret:
name: str
value: str
created_at: str = field(default_factory=lambda: datetime.now().isoformat())
expires_at: Optional[str] = None
version: int = 1
rotations: int = 0
def masked(self) -> str:
if len(self.value) <= 8:
return "****"
return self.value[:4] + "****" + self.value[-4:]
def is_expired(self) -> bool:
if not self.expires_at:
return False
return datetime.fromisoformat(self.expires_at) < datetime.now()
def days_until_expiry(self) -> Optional[int]:
if not self.expires_at:
return None
delta = datetime.fromisoformat(self.expires_at) - datetime.now()
return max(0, delta.days)
class SecretsVault:
def __init__(self, namespace: str):
self.namespace = namespace
self._store: Dict[str, Secret] = {}
self._audit_log: List[Dict] = []
def put(self, name: str, value: str, ttl_days: int = 90) -> Secret:
expires = (datetime.now() + timedelta(days=ttl_days)).isoformat()
existing = self._store.get(name)
version = (existing.version + 1) if existing else 1
rotations = (existing.rotations + (1 if existing else 0)) if existing else 0
s = Secret(name, value, expires_at=expires, version=version, rotations=rotations)
self._store[name] = s
self._log("PUT", name)
print(f" ✅ Secret '{name}' v{version} (expires in {ttl_days}d)")
return s
def get(self, name: str, requester: str = "system") -> Optional[str]:
s = self._store.get(name)
if not s:
self._log("GET_MISS", name, requester)
return None
if s.is_expired():
self._log("GET_EXPIRED", name, requester)
print(f" ⚠️ Secret '{name}' منتهي الصلاحية!")
return None
self._log("GET", name, requester)
return s.value
def rotate(self, name: str, new_value: str):
if name not in self._store:
print(f" ❌ Secret '{name}' غير موجود")
return
self.put(name, new_value)
print(f" 🔄 تم تدوير Secret '{name}' بنجاح")
def _log(self, action: str, secret: str, user: str = "system"):
self._audit_log.append({
"ts": datetime.now().strftime("%H:%M:%S"),
"action": action, "secret": secret, "user": user,
})
def health_report(self):
print(f"\n📊 حالة Vault — {self.namespace}:")
print(f" Secrets : {len(self._store)}")
expiring_soon = [s for s in self._store.values()
if (d := s.days_until_expiry()) is not None and d < 14]
expired = [s for s in self._store.values() if s.is_expired()]
print(f" تنتهي قريباً (< 14 يوم): {len(expiring_soon)}")
print(f" منتهية الصلاحية : {len(expired)}")
print(f"\n تفاصيل:")
for name, s in self._store.items():
d = s.days_until_expiry()
d_s = f"{d} يوم" if d is not None else "لا تنتهي"
icon = "🔴" if s.is_expired() else ("🟡" if d is not None and d < 14 else "🟢")
print(f" {icon} {name:<30} v{s.version} {s.masked():<15} {d_s}")
# ─── مسح الكود ─────────────────────────────────────────────
print("🔍 فحص الكود عن Secrets مُسرَّبة:")
print("=" * 52)
SAMPLE_CODE = """
import anthropic
# هذا خطأ فادح! ❌
client = anthropic.Anthropic(api_key="sk-ant-api03-xXxXxXxXxXxXxXxXxX")
AWS_KEY = "AKIAIOSFODNN7EXAMPLE"
password = "mysecretpassword123"
# الطريقة الصحيحة ✅
import os
client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
"""
scanner = SecretScanner()
findings = scanner.scan_code(SAMPLE_CODE, "app/config.py")
print(f" 📄 فحص: app/config.py")
if findings:
print(f" 🚨 وُجد {len(findings)} secret(s) مكشوفة:")
for f in findings:
print(f" ❌ [{f['type']}] {f['snippet']}")
else:
print(f" ✅ لا secrets مكشوفة")
# ─── Vault ─────────────────────────────────────────────────
print(f"\n\n🔐 Secrets Vault:")
vault = SecretsVault("production")
vault.put("ANTHROPIC_API_KEY", "sk-ant-api03-real-key-here", ttl_days=90)
vault.put("OPENAI_API_KEY", "sk-proj-openai-key-here", ttl_days=90)
vault.put("DATABASE_URL", "postgresql://user:pass@host", ttl_days=365)
vault.put("JWT_SECRET", "super-random-jwt-secret-256", ttl_days=30)
vault.put("SUPABASE_KEY", "sbp_old-key-expiring-soon", ttl_days=3) # منتهي قريباً
# قراءة Secrets
print(f"\n\n🔍 قراءة Secrets بأمان:")
for key in ["ANTHROPIC_API_KEY", "JWT_SECRET"]:
val = vault.get(key, requester="api-server")
s = vault._store.get(key)
print(f" • {key}: {s.masked() if s else 'N/A'}")
# Rotation
print(f"\n\n🔄 تدوير السر (Rotation):")
vault.rotate("JWT_SECRET", "new-super-random-jwt-secret-256-bit")
# تقرير الصحة
vault.health_report()
# Best Practices
print(f"\n\n📌 قواعد Secrets Management:")
rules = [
"❌ لا تضع Secrets في الكود أو Git أبداً",
"✅ استخدم env vars محلياً (.env + .gitignore)",
"✅ استخدم Vault أو Secrets Manager في الإنتاج",
"✅ دوّر الـ Secrets كل 90 يوم",
"✅ راجع Audit Log بشكل دوري",
"✅ استخدم git-secrets لمنع الإضافة الخاطئة",
]
for r in rules:
print(f" {r}")
print(f"\n✅ Secrets Management — دورة أمن السحاب مكتملة!")