Dacă ați petrecut ceva timp construind agenți IA pregătiți pentru producție, probabil că v-ați confruntat cu o mare problemă de securitate legată de accesul la instrumente agentice. În arhitectura software clasică, avem limite clare și statice care separă codul executabil de datele brute introduse de utilizator. Dar atunci când construiți aplicații în jurul unui LLM, aceste limite devin inutile. LLM-urile procesează prompturile sistemului, contextul dinamic preluat și intrările nesigure ale utilizatorului în exact aceeași fereastră de context. Acest design face ca sistemele să fie extrem de vulnerabile la injectarea de prompturi: un atacator poate crea cu ușurință text de intrare care suprascrie instrucțiunile sistemului și deturnează motorul de raționament al modelului.
Această criză de încredere devine periculoasă atunci când doriți, spre exemplu, să dați acces unui LLM direct la componente de sistem într-o arhitectură de cloud prin intermediul unui MCP. Dacă serverul MCP expune o suită standard de instrumente de diagnoză, brute (cum ar fi ping_node, fetch_routing_table, sau restart_service) pentru gestiunea acțiunilor de asigurare a stabilității unei aplicații în cloud, vă confruntați cu o dublă problemă:
Aceasta este problema clasică de confuzie de autor: agentul acționează cu autoritatea deplină a serverului backend, dar execută o intenție malițioasă transmisă de un utilizator neautorizat.
Soluția este de a schimba complet granița de securitate. În loc să permitem LLM-ului să execute cod arbitrar sau să orchestreze zeci de taskuri granulare de diagnostic, expunem un singur task extrem de constrâns, care acceptă un program scris într-o gramatică formală a unui DSL de micro-orchestrare. LLM-ul trimite un mini-script de diagnosticare cu instrucțiuni de ramificare explicite și predefinite. Serverul MCP la distanță analizează acest script, verifică conformitatea sa de securitate în raport cu domeniile de aplicare OAuth 2.1 validate ale utilizatorului și execută întregul bloc de diagnosticare în siguranță într-o singură tranzacție locală, de ordinul milisecundelor.
În loc să permitem LLM-ului să coordoneze acțiuni complexe ale sistemului prin apeluri API separate, implementăm un strat de execuție constrâns gramatical. Ciclul de viață decuplează raționamentul în limbaj natural al LLM de interacțiunea directă cu mediul:
Hand-shake pentru identitate: Clientul stabilește o conexiune securizată și cu stare completă cu serverul MCP la distanță utilizând OAuth 2.1.
Descoperire dinamică: Clientul solicită lista de instrumente. Serverul își promovează instrumentul de micro-orchestrare și publică gramatica formală EBNF/GBNF care definește sintaxa admisă a DSL-ului de diagnosticare.
Generare ghidată: Motorul gazdă constrânge procesul de generare a tokenurilor de către LLM folosind generarea ghidată de gramatică, asigurându-se că modelul este incapabil fizic să emită tokenuri care încalcă sintaxa DSL.
Analiza deterministă a serverului: Clientul trimite sarcina utilă a programului DSL generat către serverul MCP la distanță, care o procesează folosind un parser recursive-descent pentru a construi un arbore de sintaxă abstractă (AST) puternic tipizat.
Un interpretor DSL securizat este inutil dacă clienții neautorizați pot accesa mediul de execuție. Deoarece serverele MCP la distanță operează prin transporturi bazate pe HTTP, acestea trebuie să respecte standardele formale de autorizare definite în specificația MCP, care se aliniază strict cu OAuth 2.1.
Domeniile de autorizare OAuth nu sunt doar pentru blocarea punctelor finale - în arhitectura noastră, acestea modelează dinamic gramatica expusă modelului și guvernează modul în care este evaluat AST. Am împărțit această abordare de apărare în profunzime în două straturi cheie:
Când clientul interoghează lista de instrumente de la serverul MCP, serverul inspectează domeniile de autorizare ale tokenurilor OAuth active ale apelantului. În loc să ofere exact aceeași gramatică tuturor, serverul elimină dinamic din schema EBNF din metadate instrucțiuni gramaticale bazat pe domeniile de aplicare specifice ale utilizatorului.
De exemplu, dacă un utilizator are doar mcp:diag:read în domeniul de aplicare, gramatica returnată va defini doar cuvinte cheie care nu sunt generatoare de mutații, cum ar fi ping, get_status și if_running. Motorul de generare ghidată al LLM este limitat fizic și nu poate genera cuvinte cheie care generează modificări, cum ar fi restart, deoarece acestea nu există în gramatica activă.
Dacă un atacator ocolește constrângerile de pe partea clientului, interpretorul de pe partea serverului acționează ca o limită rigidă de tip zero-trust. Fiecare nod din AST-ul analizat este verificat în raport cu domeniile de aplicare validate ale utilizatorului înainte de orice execuție:
Citiri de diagnosticare: Operațiuni precum ping și get_status necesită ca domeniu de aplicare mcp:diag:read. Dacă tokenul purtător nu are acest domeniu, interpretorul respinge operația și generează o eroare.
reset necesită în domeniul de aplicare mcp:diag:write. O încălcare oprește execuția și returnează o provocare de tip "step-up" care conține domeniile de aplicare necesare.F# este un limbaj incredibil de puternic pentru scrierea de compilatoare și interpretoare sigure, datorită sistemului său strict de tipuri și imutabilității. Pentru a analiza DSL-ul nostru securizat de micro orchestrare, putem folosi FParsec, o librărie de parser combinators matură și extrem de optimizată, construită special pentru F#.
FParsec are o abordare de sus în jos, recursive-descent. Aceasta înseamnă că reușim să construim parsere complexe prin compunerea unor funcții modulare mici, folosind operatori de combinare clari. Librăria acceptă nativ gramaticile LL(k) și LL() cu look-ahead arbitrar, și revenirea în stream* folosind operatorul attempt atunci când o ramură eșuează.
Deoarece FParsec folosește recursive -descent, trebuie să țineți cont de două aspecte importante de proiectare:
Recursivitate la stânga: Parserele recursive-descent vor intra într-o buclă infinită și se vor bloca cu un overflow al stivei dacă întâlnesc recursivitate directă la stânga (unde o regulă se referă la ea însăși ca prim simbol). Trebuie să restructurați regulile gramaticale într-o formă recursivă la dreapta sau să utilizați componenta specializată de analiză a precedenței operatorilor din FParsec pentru a gestiona ierarhiile de expresii în siguranță.
parse { let! x = p... }), utilizarea lor în fluxul principal de analiză introduce o supraîncărcare semnificativă a alocării pe stack de closures. Pentru performanțe maxime și o presiune mai mică asupra garbage-collector-ului, ar trebui să scrieți parserii folosind operatorii de combinare predefiniți ai FParsec, cum ar fi >>., .>>, .>>. și |>>.Iată cum arată o implementare completă, type-safe, scrisă în F# funcțional pur, folosind librăria FParsec. Începem prin a defini tipurile de date ale domeniului, claims pentru user și AST-ul nostru de micro orchestrare (care acceptă liste de comenzi, recuperarea din erori în try-catch și ramificarea condiționată) folosind uniunile discriminate și tipurile de date din F#:
namespace SecureMcp
open FParsec
type Scope =
| DiagRead
| DiagWrite
type UserClaims = {
UserId: string
Scopes: Scope list
}
type Command =
| Ping of host: string
| GetStatus of service: string
| RestartService of service: string
type DiagAST =
| Action of Command
| Sequence of DiagAST list
| TryCatch of tryBlock: DiagAST * catchBlock: DiagAST
| IfRunning of service: string * thenBlock: DiagAST * elseBlock: DiagAST
type ExecutionError =
| UnauthorizedTable of string
| FieldAccessDenied of field: string * value: string
| InsufficientScope of requiredScope: string
| ParserFailure of stringApoi, scriem parser-combinatorul folosind sintaxa optimizată de pipeline de operatori a FParsec. Folosim createParserForwardedToRef pentru a gestiona în mod clar definițiile recursive necesare pentru ramificarea condiționată și blocurile try-catch:
let strLiteral : Parser =
pchar '"' >>. manyCharsTill anyChar (pchar '"')
let ws = spaces
// Base diagnostic action parsers
let pingParser : Parser =
pstring "PING" >>. ws >>. strLiteral |>>
(fun host -> Action (Ping host))
let statusParser : Parser =
pstring "GET_STATUS" >>. ws >>. strLiteral
|>> (fun svc -> Action (GetStatus svc))
let restartParser : Parser =
pstring "RESTART" >>. ws >>. strLiteral |>>
(fun svc -> Action (RestartService svc))
let actionParser : Parser =
choice [ attempt pingParser; attempt
statusParser; restartParser ]
// Setup forward reference for recursive
// branching rules
let expr, exprImpl =
createParserForwardedToRef()
// Parses: IF_RUNNING "service" THEN {... } ELSE {... }
let ifRunningParser : Parser =
pstring "IF_RUNNING" >>. ws >>. strLiteral.>> ws
.>> pstring "THEN".>> ws.>>. expr.>> ws
.>> pstring "ELSE".>> ws.>>. expr
|>> fun ((svc, thenB), elseB) ->
IfRunning(svc, thenB, elseB)
// Parses: TRY {... } CATCH {... }
let tryCatchParser : Parser =
pstring "TRY" >>. ws >>. expr.>> ws.>>
pstring "CATCH".>> ws.>>. expr
|>> fun (tryB, catchB) -> TryCatch(tryB,
catchB)
// Parses block wrappers: { cmd1 cmd2 }
let blockParser : Parser =
between (pchar '{'.>> ws) (ws.>> pchar '}')
(sepBy expr ws)
|>> Sequence
// Resolve the recursive parser implementation
do exprImpl := choice
// Complete program entry point
let diagParser : Parser = ws >>.
expr.>> ws.>> eof Într-un final, implementăm interpretorul server-side. Acesta evaluează recursiv AST-ul de diagnostic, verificând domeniile de autorizare ale identității validate ale apelantului la fiecare nod și returnează un jurnal secvențial al operațiunilor:
module Interpreter =
type Env = {
RunningServices: string list
PingResult: string -> bool
}
let rec eval (ast: DiagAST) (claims: UserClaims) (env: Env) : Result =
match ast with
| Action cmd ->
match cmd with
| Ping host ->
if not (List.contains DiagRead claims.Scopes) then
Error (InsufficientScope "mcp:diag:read")
else
let ok = env.PingResult host
Ok
| GetStatus svc ->
if not (List.contains DiagRead claims.Scopes) then
Error (InsufficientScope "mcp:diag:read")
else
let running = List.contains svc env.
RunningServices
Ok
| RestartService svc ->
if not (List.contains DiagWrite claims.Scopes) then
Error (InsufficientScope "mcp:diag:write")
else
Ok
| Sequence asts ->
let rec evalSeq acc remaining =
match remaining with
| -> Ok (List.concat (List.rev acc))
| head :: tail ->
match eval head claims env with
| Error err -> Error err
| Ok logs -> evalSeq (logs :: acc) tail
evalSeq asts
| TryCatch (tryB, catchB) ->
match eval tryB claims env with
| Ok logs -> Ok logs
| Error _ ->
// Attempt automatic containment/recovery
//locally on the server
match eval catchB claims env with
| Ok catchLogs -> Ok ( "CONTAINMENT_TRIGGERED"
:: catchLogs )
| Error err -> Error err
| IfRunning (svc, thenB, elseB) ->
if not (List.contains DiagRead claims.Scopes) then
Error (InsufficientScope "mcp:diag:read")
else
let running = List.contains svc env.RunningServices
if running then
eval thenB claims env
else
eval elseB claims env Comunicarea dintre aplicația gazdă, clientul MCP și serverul la distanță respectă protocolul structurat de mesaje JSON-RPC 2.0. Iată o hartă completă a fluxului care prezintă modul în care se execută descoperirea capabilităților și apelurile de instrumente securizate:
Mesajele JSON-RPC structurate fac ca aceste limite de comunicare să fie clare și standardizate:
Clientul solicită o listă de instrumente disponibile de la server:
{
"jsonrpc": "2.0",
"id": "req_001",
"method": "tools/list",
"params": {}
}Serverul răspunde, publicând toolul execute_diagnostic_pipeline expunând schema gramaticală trunchiată, indicând că acceptă doar șiruri de caractere care corespund sintaxei noastre de execuție:
{
"jsonrpc": "2.0",
"id": "req_001",
"result": {
"resultType": "complete",
"tools":,
"additionalProperties": false
}
}
]
}
}Modelul gazdă scrie un program recursiv conceput pentru a verifica un serviciu, a da un ping unui endpoint API și a declanșa un bloc de repornire dacă oricare dintre etape eșuează:
{
"jsonrpc": "2.0",
"id": "req_002",
"method": "tools/call",
"params": {
"name": "execute_diagnostic_pipeline",
"arguments": {
"program": "TRY { IF_RUNNING \"database\" THEN { PING \"api.internal\" } ELSE { RESTART \"database\" } } CATCH { RESTART \"database\" }"
}
}
}Serverul analizează, interpretează și evaluează pipeline-ul într-o singură tranzacție cu datele disponibile. Deoarece tokenul utilizatorului conține ambele domenii de autorizare diag:read și diag:write, programul se execută și returnează un rezultat pozitiv:
{
"jsonrpc": "2.0",
"id": "req_002",
"result": {
"resultType": "complete",
"content":"
}
]
}
}Dacă LLMul emite un program care încearcă să execute o mutație restart, dar tokenul utilizatorului nu are în domeniul de autorizare mcp:diag:write, interpretorul de pe server oprește execuția la acel nod specific:
{
"jsonrpc": "2.0",
"id": "req_003",
"result": {
"isError": true,
"content": [
{
"type": "text",
"text": "Execution rejected: operation requires scope 'mcp:diag:write'"
}
]
}
}Consolidarea securității agenților bazați pe LLMuri necesită mutarea graniței de încredere de la prompting nedeterminist în limbaj natural la abordări bazate pe teoria clasică a compilatoarelor.
Prin constrângerea interfeței LLM-ului printr-un DSL scris într-o gramatică formală se diminuează matematic întreaga suprafață de atac disponibilă. Chiar dacă un atacator reușește o injecție de prompturi excelentă, progresul acestuia este complet limitat de către sintaxa DSL-ului. Sintaxa F# strictă și semantica clară de interpretare bazată pe parser-combinators din FParsec contribuie la soliditatea absolută a oricărei implementări de gramatici cu sau fără trunchiere de instrucțiuni ghidate de domenii de autorizare OAuth. Acest lucru vă oferă un framework compatibil cu conceptele de cod auditabil, zero-trust, și pregătit pentru IA în producție.
de Vlad Petrean
de Ovidiu Mățan
