لماذا لا ينبغي إعطاء المفتاح الرئيسي لعميل Node.js وكيفية تنفيذ رمز مؤقت مدته 60 ثانية
عند بناء وكلاء ذاتيين (Autonomous Agents) باستخدام LangChain أو LlamaIndex، تصل حتمًا إلى لحظة ربطهم بأساس البيانات (DB) وواجهات برمجة التطبيقات الخارجية (APIs). وهنا عادةً ما تقع الحوادث. فضربة واحدة من الحقن البرمجي للموجهات (Prompt Injection) كفيلة بكشف مفتاح OpenAI الرئيسي أو كلمة مرور مسؤول قاعدة البيانات المخزنة في ملف .env. ولا فائدة تُرجى من إحكام حواجز الحماية للموجهات مهما بلغت دقتها؛ فما دامت منطقة الاستدلال منطقياً مرتبطة بمنطقة التنفيذ في نماذج اللغات الكبيرة (LLM)، سينهار الأمان بسبب جملة واحدة.
وفقًا لتقرير إطار عمل الذكاء الاصطناعي الآمن (SAIF) لعام 2025 الصادر عن Google، تعرضت 88% من الشركات التي تبنت الوكلاء الأذكياء لمحاولات حقن الموجهات. وكانت نسبة المنع باستخدام تقنيات اكتشاف أنماط النصوص التقليدية لا تتجاوز 23%. لذلك، من الأفضل أن تريح بالك وألا تثق بفرآيند الوكيل نفسه على الإطلاق. لا تمنح الوكيل أي صلاحيات مطلقاً، بل قم بتغيير البنية التحتية لحقن رمز مؤقت يعمل لمدة 60 ثانية فقط في طبقة البرمجيات الوسيطة (Middleware Layer).
تنفيذ حقن رمز Vault الديناميكي لمدة 60 ثانية في Express
باستخدام مصادقة AppRole من HashiCorp Vault، يمكنك إصدار رمز ذي عمر 60 ثانية فقط في اللحظة التي يتم فيها استدعاء أداة معينة. عندما يطلب الوكيل API خارجيًا، يتداخل المعترض (Interceptor) في منتصف العملية ليدفع بالرمز قصير العمر (Short-lived token) داخل الترويسة (Header).
`typescript
import { Request, Response, NextFunction } from 'express';
import vault from 'node-vault';
interface VaultAppRoleAuth {
roleId: string;
secretId: string;
}
export class EphemeralTokenInjector {
private vaultClient: any;
private roleId: string;
private secretId: string;
constructor(endpoint: string, auth: VaultAppRoleAuth) {
this.vaultClient = vault({ endpoint });
this.roleId = auth.roleId;
this.secretId = auth.secretId;
}
private async getAppRoleToken(): Promise {
const result = await this.vaultClient.approleLogin({
role_id: this.roleId,
secret_id: this.secretId,
});
return result.auth.client_token;
}
public createToolInterceptor(targetServiceRole: string) {
return async (req: Request, res: Response, next: NextFunction) => {
let clientToken: string | null = null;
try {
clientToken = await this.getAppRoleToken();
const dynamicSecret = await this.vaultClient.write(
`sys/leases/generate/${targetServiceRole}`,
{ ttl: '60s' }
);
req.headers['authorization'] = `Bearer ${dynamicSecret.data.token}`;
req.body.ephemeralContext = {
leaseId: dynamicSecret.lease_id,
expiresAt: Date.now() + 60000,
};
next();
} catch (error) {
res.status(500).json({ error: 'Failed to inject ephemeral dynamic secret' });
} finally {
clientToken = null;
}
};
}
}
`
لكن كتابة الكود وحدها لا تكفي. نظرًا لأن ذاكرة التكديس (Heap Memory) لـ Node.js V8 يعمل فيها جامع القمامة (Garbage Collector) بشكل عشوائي، فإن سلسة النص للرمز تظل موجودة في الذاكرة لفترة حتى بعد تفريغ المتغيرات. وفي حالة التعرض لهجوم تفريغ الذاكرة (Heap Dump)، هناك خطر التسريب.
- بمجرد استلام استجابة الAPI، ضع القيم الحساسة فورًا داخل كائن
Buffer.
- بعد الانتهاء من التنفيذ، قم بتشغيل
Buffer.fill(0) لتطهير البايتات قسريًا وتحويلها إلى 0.
- قم بتعيين
null لمتغير المرجع ليمر إلى جامع القمامة (GC).
باتباع هذه الخطوات الثلاث فقط، تقل احتمالية تسريب الرمز عبر تحليل الذاكرة (Memory Profiling) بشكل كبير.
| طريقة إدارة بيانات الاعتماد |
متوسط العمر (TTL) |
نطاق الضرر عند الاختراق |
تتبع التدقيق |
| Master API Key المدمج بالكود |
غير محدود |
الاستحواذ الكامل على صلاحيات البنية التحتية |
تعذر تحديد الهوية بسبب مشاركة مفتاح واحد |
| Static Secret في متغيرات البيئة (.env) |
عمر عملية البرمجية |
إساءة استخدام كامل صلاحيات عملية Node المعنية |
تعذر تتبع مسارات عمل المستخدمين المفردين |
| Vault AppRole Ephemeral Token |
60 ثانية |
مقتصر على تنفيذ الأداة المعنية لمرة واحدة |
تتبع بالثواني عبر Vault Audit Log |
تمرير Auth0 OIDC Claim إلى سياسة PostgreSQL RLS
حتى لو نفذ الوكيل أمر "اعرض لي جميع بيانات الأعضاء" نتيجة لاختراق الموجهات (Prompt Jailbreak)، فيكفي أن يرفض محرك قاعدة البيانات هذا الطلب. إذا كنت تعمل في بيئة SaaS متعددة المستأجرين (Multi-tenant)، فيجب عليك إعداد أمان مستوى الصفوف (Row Level Security - RLS) في PostgreSQL.
احصل على userId و tenantId من JWT الصادر عن Auth0 ومررها إلى سياق جلسة RunnableConfig في LangChain. ثم احقن هذا السياق كمتغير جلسة داخل معاملة Prisma (Transaction).
`typescript
import { RunnableConfig } from '@langchain/core/runnables';
import { PrismaClient } from '@prisma/client';
export interface AgentUserClaims {
userId: string;
tenantId: string;
}
export async function executeAgentToolWithRLS(
prisma: PrismaClient,
config: RunnableConfig,
dbOperation: (tx: any) => Promise
): Promise {
const claims = config.configurable?.userClaims as AgentUserClaims;
if (!claims || !claims.tenantId || !claims.userId) {
throw new Error('Unauthorized: Missing OIDC Claims in Agent Execution Context');
}
return await prisma.transaction(async (tx) => {
await tx.executeRawSELECT set_config('app.current_tenant_id', ${claims.tenantId}, true);
await tx.$executeRawSELECT set_config('app.current_user_id', ${claims.userId}, true);
return await dbOperation(tx);
});
}
`
يجب التمرير بـ true كمعامل أخير لـ set_config لضمان تطبيق المتغير فقط على نطاق المعاملة الحالية (SET LOCAL). وهذا إعداد رئيسي لمنع تسرب صلاحيات المستخدم السابق إلى الطلب التالي في بيئات تجميع اتصالات قاعدة البيانات (DB Connection Pooling).
الان أنشئ سياسة تقرأ متغير الجلسة هذا في ملف SQL.
`sql
ALTER TABLE tenant_documents ENABLE ROW LEVEL SECURITY;
ALTER TABLE tenant_documents FORCE ROW LEVEL SECURITY;
CREATE POLICY agent_tenant_isolation_policy ON tenant_documents
FOR ALL
TO authenticated_agent_role
USING (
tenant_id = current_setting('app.current_tenant_id', true)::uuid
)
WITH CHECK (
tenant_id = current_setting('app.current_tenant_id', true)::uuid
);
`
بهذه التوليفة، مهما أنشأ الوكيل استعلامات غريبة، فإنه لن يستطيع حتى الاستعلام عن البيانات الخارجة عن نطاق المستأجر الخاص به. وحتى لو حدث تسريب، يمكن تقليل وقت التعافي من الكوارث (MTTR) من عدة أيام إلى بضع دقائق فقط.
خط أنابيب التحقق في CI/CD باستخدام Promptfoo و PyRIT
لا يمكن للمطورين اختبار عزل الصلاحيات يدويًا في كل مرة يغيرون فيها كود أدوات الوكيل. نربط الأدوات مفتوحة المصدر مثل Promptfoo و PyRIT من Microsoft مع GitHub Actions للتحقق مع كل طلب سحب (PR).
`yaml
name: Agent Red Teaming Security Gate
on:
pull_request:
paths:
- 'src/agents/'
- 'src/tools/'
- 'prompts/**'
jobs:
security-eval:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Setup Node.js
uses: actions/setup-node@v4
with:
node-version: 20
cache: 'npm'
- name: Install Dependencies
run: npm ci
- name: Run Promptfoo Scan
uses: promptfoo/promptfoo-action@v1
with:
config: 'promptfooconfig.yaml'
openai-api-key: ${{ secrets.OPENAI_API_KEY }}
github-token: ${{ secrets.GITHUB_TOKEN }}
- name: Run PyRIT Multi-Turn Test
env:
AGENT_ENDPOINT: 'http://localhost:3000/api/agent'
run: |
python -m pip install pyrit
python scripts/run_pyrit_eval.py --endpoint $AGENT_ENDPOINT --pass-threshold 0.98
`
في بيئة تشغيل الخادم، نربط برمجية وسيطة لقاطع الدائرة (Circuit Breaker Middleware) تقوم بإيقاف التشغيل فورًا عند اكتشاف أنماط التهديد.
`typescript
import { Request, Response, NextFunction } from 'express';
export class AgentCircuitBreaker {
private failureCount: number = 0;
private readonly threshold: number = 3;
private state: 'CLOSED' | 'OPEN' | 'HALF_OPEN' = 'CLOSED';
private forbiddenSignatures: RegExp[] = [
/ignore\s+all\s+previous\s+instructions/i,
/system::override_privileges/i,
/grant\s+role\s+admin/i,
/concat\s*(\s*select/i
];
public middleware() {
return (req: Request, res: Response, next: NextFunction) => {
if (this.state === 'OPEN') {
return res.status(503).json({
error: 'CircuitBreaker:Open - Agent execution halted'
});
}
const promptInput = JSON.stringify(req.body);
const isPatternViolated = this.forbiddenSignatures.some(sig => sig.test(promptInput));
if (isPatternViolated) {
this.failureCount++;
this.dispatchSecurityAlert(req.body);
if (this.failureCount >= this.threshold) {
this.state = 'OPEN';
}
return res.status(403).json({
error: 'Security Policy Violation: Malicious prompt pattern'
});
}
next();
};
}
private dispatchSecurityAlert(payload: any): void {
const webhookUrl = process.env.SECURITY_WEBHOOK_URL;
if (!webhookUrl) return;
fetch(webhookUrl, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({
event: 'AGENT_PRIVILEGE_ESCALATION_DETECTED',
timestamp: new Date().toISOString(),
payload
})
}).catch(() => {});
}
}
`
خطوات بناء خط أنابيب اختبار الأمان بسيطة:
- ضع
promptfooconfig.yaml في جذر المشروع وحدد عناصر الفحص لـ jailbreak و excessive-agency.
- أنشئ سير عمل اكتشاف PR في GitHub Actions لتشغيل سيناريوهات Promptfoo و PyRIT.
- قم بوضع البرمجية الوسيطة
AgentCircuitBreaker أمام نقطة نهاية Express لإيقاف تنفيذ الوكيل عند اكتشاف أنماط الحقن 3 مرات متتالية.
باستخدام هذا الإعداد، يمكنك توفير أكثر من 5 ساعات كانت تُقضى أسبوعيًا في مراجعة الموجهات يدويًا.
| مؤشر التقييم |
طريقة التحقق اليدوي |
إدخال الأتمتة ذات الثقة الصفرية (Zero-Trust) |
| وقت الإعداد للتدقيق |
5 إلى 8 ساعات أسبوعيًا |
أقل من ساعة واحدة أسبوعيًا |
| MTTR عند تسرب الصلاحيات |
عدة أيام (فحص شامل لجميع السجلات/قاعدة البيانات) |
بضع دقائق (مقتصر على نطاق RLS) |
| معدل منع حقن الموجهات |
حوالي 23% |
99.9% |
| مخاطر الفوترة عند تسرب المفتاح |
فوترة سحابية غير محدودة |
منع الفوترة بفضل TTL البالغ 60 ثانية |
الجوهر في أمان الوكلاء الأذكياء ليس بالدعاء أن يطيع النموذج الأوامر، بل بوضع القيود والأغلال على مستوى البنية التحتية بحيث لا يستطيع النموذج إلحاق الضرر بالنظام حتى لو قال كلاماً فارغاً أو سقط في فخ هجوم.