qefro_backend_sdk/
lib.rs

1use std::collections::HashMap;
2use std::sync::Arc;
3
4use anyhow::Result;
5use async_trait::async_trait;
6use chrono::Utc;
7use hmac::{Hmac, Mac};
8use serde::{Deserialize, Serialize};
9use serde_json::{json, Value};
10use sha2::Sha256;
11use uuid::Uuid;
12
13type HmacSha256 = Hmac<Sha256>;
14
15/// Package name reported to Qefro Runtime (`X-Qefro-SDK` / protocol payloads).
16pub const SDK_NAME: &str = "qefro-backend-sdk";
17/// Package version reported to Qefro Runtime (`sdk_version` / `X-Qefro-Version`).
18pub const SDK_VERSION: &str = env!("CARGO_PKG_VERSION");
19
20type ToolHandler = Arc<dyn Fn(ToolContext) -> ToolFuture + Send + Sync>;
21type ToolFuture = std::pin::Pin<Box<dyn std::future::Future<Output = Result<Value>> + Send>>;
22
23#[derive(Debug, Clone)]
24pub struct QefroConfig {
25    pub signing_secret: String,
26    pub protocol_version: String,
27    pub max_timestamp_skew_secs: i64,
28}
29
30impl QefroConfig {
31    pub fn new(signing_secret: impl Into<String>) -> Self {
32        Self {
33            signing_secret: signing_secret.into(),
34            protocol_version: "1".into(),
35            max_timestamp_skew_secs: 300,
36        }
37    }
38}
39
40#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
41#[serde(rename_all = "snake_case")]
42pub enum ToolAuthMode {
43    None,
44    Optional,
45    Required,
46}
47
48impl Default for ToolAuthMode {
49    fn default() -> Self {
50        Self::Optional
51    }
52}
53
54#[derive(Debug, Clone, Serialize, Deserialize, Default)]
55pub struct ToolMetadata {
56    pub name: String,
57    #[serde(default)]
58    pub description: Option<String>,
59    #[serde(default)]
60    pub input_schema: Option<Value>,
61    #[serde(default)]
62    pub authentication_methods: Vec<String>,
63    #[serde(default)]
64    pub auth: ToolAuthMode,
65    #[serde(default)]
66    pub permissions: Vec<String>,
67    #[serde(default)]
68    pub timeout: Option<u64>,
69    #[serde(default)]
70    pub default_auth_method: Option<String>,
71}
72
73#[derive(Debug, Clone, Serialize, Deserialize)]
74pub struct RegisteredTool {
75    pub name: String,
76    #[serde(default)]
77    pub description: Option<String>,
78    #[serde(default)]
79    pub input_schema: Option<Value>,
80    #[serde(default)]
81    pub authentication_methods: Vec<String>,
82    #[serde(default)]
83    pub auth: ToolAuthMode,
84    #[serde(default)]
85    pub permissions: Vec<String>,
86    #[serde(default)]
87    pub timeout: Option<u64>,
88}
89
90#[derive(Debug, Clone, Serialize, Deserialize)]
91pub struct AuthenticationContextPayload {
92    #[serde(rename = "type")]
93    pub credential_type: Option<String>,
94    pub access_token: Option<String>,
95    pub credential: Option<String>,
96    pub refresh_token: Option<String>,
97    pub expires_in: Option<i64>,
98    pub customer_id: Option<String>,
99}
100
101#[derive(Debug, Clone, Serialize, Deserialize)]
102pub struct ChallengePayload {
103    #[serde(rename = "type")]
104    pub challenge_type: String,
105    pub message: String,
106    pub destination_hint: Option<String>,
107    pub login_url: Option<String>,
108}
109
110#[derive(Debug, Clone, Serialize, Deserialize)]
111pub struct QefroRequest {
112    pub protocol_version: String,
113    pub request_id: Uuid,
114    #[serde(rename = "type")]
115    pub request_type: String,
116    pub organization_id: Option<Uuid>,
117    pub conversation_id: Option<Uuid>,
118    pub channel: Option<String>,
119    pub identity: Option<Value>,
120    pub tool: Option<String>,
121    pub parameters: Option<Value>,
122    pub authentication: Option<Value>,
123    pub resume_token: Option<String>,
124    pub challenge_response: Option<String>,
125}
126
127#[derive(Debug, Clone, Serialize, Deserialize)]
128#[serde(tag = "type", rename_all = "snake_case")]
129pub enum QefroResponse {
130    Pong {
131        protocol_version: String,
132        sdk_version: String,
133    },
134    #[serde(rename = "tools.list")]
135    ToolsList {
136        tools: Vec<RegisteredTool>,
137        protocol_version: String,
138        sdk_version: String,
139    },
140    Result {
141        output: Value,
142        authentication_context: Option<AuthenticationContextPayload>,
143    },
144    Challenge {
145        resume_token: String,
146        challenge: ChallengePayload,
147    },
148    Error {
149        code: String,
150        message: String,
151    },
152}
153
154#[derive(Debug, Clone)]
155struct PendingInvocation {
156    tool: String,
157    conversation_id: Uuid,
158    parameters: Value,
159    identity: Option<Value>,
160    channel: Option<String>,
161}
162
163#[derive(Debug, Clone)]
164struct StoredAuth {
165    customer: Value,
166    auth: AuthenticationContextPayload,
167    expires_at_epoch_ms: i64,
168}
169
170#[derive(Clone)]
171pub struct ToolContext {
172    pub identity: Value,
173    pub parameters: Value,
174    pub conversation_id: Uuid,
175    pub channel: Option<String>,
176    pub authentication: Option<Value>,
177    pub auth_response: Option<String>,
178    pub customer: Option<Value>,
179}
180
181#[derive(Debug, Clone)]
182pub enum AuthOutcome {
183    Success {
184        customer: Value,
185        auth: AuthenticationContextPayload,
186    },
187    Challenge(ChallengePayload),
188    Denied,
189    NotFound,
190}
191
192#[derive(Debug, Clone)]
193pub struct CustomerLookupContext {
194    pub identity: Value,
195    pub parameters: Value,
196    pub conversation_id: Uuid,
197    pub channel: Option<String>,
198}
199
200#[derive(Debug, Clone)]
201pub struct CustomerAuthorizeContext {
202    pub customer: Value,
203    pub method: Option<String>,
204    pub response: Option<String>,
205    pub identity: Value,
206    pub parameters: Value,
207    pub conversation_id: Uuid,
208    pub channel: Option<String>,
209}
210
211#[async_trait]
212pub trait CustomerProvider: Send + Sync {
213    async fn lookup(&self, ctx: &CustomerLookupContext) -> Result<Option<Value>>;
214    async fn authorize(&self, ctx: &CustomerAuthorizeContext) -> Result<AuthOutcome>;
215}
216
217#[derive(Clone)]
218struct ToolRegistration {
219    metadata: ToolMetadata,
220    handler: ToolHandler,
221}
222
223#[derive(Debug, Clone)]
224pub struct ListenOptions {
225    pub port: u16,
226    pub host: Option<String>,
227    pub path: Option<String>,
228}
229
230#[derive(Debug, Clone)]
231pub struct ListenHandle {
232    pub url: String,
233}
234
235pub struct Qefro {
236    config: QefroConfig,
237    tools: HashMap<String, ToolRegistration>,
238    pending: HashMap<String, PendingInvocation>,
239    auth_by_conversation: HashMap<Uuid, StoredAuth>,
240    customer_provider: Option<Arc<dyn CustomerProvider>>,
241}
242
243impl Qefro {
244    pub fn new(config: QefroConfig) -> Self {
245        Self {
246            config,
247            tools: HashMap::new(),
248            pending: HashMap::new(),
249            auth_by_conversation: HashMap::new(),
250            customer_provider: None,
251        }
252    }
253
254    pub fn customer<P>(&mut self, provider: P) -> &mut Self
255    where
256        P: CustomerProvider + 'static,
257    {
258        self.customer_provider = Some(Arc::new(provider));
259        self
260    }
261
262    pub fn tool<F, Fut>(&mut self, metadata: ToolMetadata, handler: F)
263    where
264        F: Fn(ToolContext) -> Fut + Send + Sync + 'static,
265        Fut: std::future::Future<Output = Result<Value>> + Send + 'static,
266    {
267        let name = metadata.name.clone();
268        self.tools.insert(
269            name,
270            ToolRegistration {
271                metadata,
272                handler: Arc::new(move |ctx| Box::pin(handler(ctx))),
273            },
274        );
275    }
276
277    pub async fn listen(&self, options: ListenOptions) -> Result<ListenHandle> {
278        let host = options.host.unwrap_or_else(|| "0.0.0.0".to_string());
279        let path = options.path.unwrap_or_else(|| "/qefro".to_string());
280        Ok(ListenHandle {
281            url: format!("http://{host}:{}{path}", options.port),
282        })
283    }
284
285    pub fn verify_signature(&self, signature: &str, timestamp: i64, body: &str) -> bool {
286        let now = Utc::now().timestamp();
287        if (now - timestamp).abs() > self.config.max_timestamp_skew_secs {
288            return false;
289        }
290        let payload = format!("v1:{timestamp}:{body}");
291        let mut mac = HmacSha256::new_from_slice(self.config.signing_secret.as_bytes())
292            .expect("HMAC accepts any key length");
293        mac.update(payload.as_bytes());
294        let expected = format!("v1={}", hex::encode(mac.finalize().into_bytes()));
295        expected == signature
296    }
297
298    pub async fn handle(&mut self, request: QefroRequest) -> QefroResponse {
299        if request.protocol_version != self.config.protocol_version {
300            return QefroResponse::Error {
301                code: "protocol_mismatch".into(),
302                message: "Unsupported protocol version".into(),
303            };
304        }
305
306        match request.request_type.as_str() {
307            "ping" => QefroResponse::Pong {
308                protocol_version: self.config.protocol_version.clone(),
309                sdk_version: SDK_VERSION.into(),
310            },
311            "tools.list" => QefroResponse::ToolsList {
312                tools: self
313                    .tools
314                    .values()
315                    .map(|r| RegisteredTool {
316                        name: r.metadata.name.clone(),
317                        description: r.metadata.description.clone(),
318                        input_schema: r.metadata.input_schema.clone(),
319                        authentication_methods: r.metadata.authentication_methods.clone(),
320                        auth: r.metadata.auth,
321                        permissions: r.metadata.permissions.clone(),
322                        timeout: r.metadata.timeout,
323                    })
324                    .collect(),
325                protocol_version: self.config.protocol_version.clone(),
326                sdk_version: SDK_VERSION.into(),
327            },
328            "tool.invoke" => {
329                self
330                    .invoke(
331                        request.tool,
332                        request.parameters.unwrap_or_else(|| json!({})),
333                        request.conversation_id.unwrap_or_else(Uuid::new_v4),
334                        request.identity,
335                        request.channel,
336                        request.authentication,
337                        None,
338                    )
339                    .await
340            }
341            "tool.resume" => {
342                let Some(resume_token) = request.resume_token else {
343                    return QefroResponse::Error {
344                        code: "invalid_request".into(),
345                        message: "resume_token is required".into(),
346                    };
347                };
348                let Some(challenge_response) = request.challenge_response else {
349                    return QefroResponse::Error {
350                        code: "invalid_request".into(),
351                        message: "challenge_response is required".into(),
352                    };
353                };
354                let Some(pending) = self.pending.remove(&resume_token) else {
355                    return QefroResponse::Error {
356                        code: "not_found".into(),
357                        message: "resume token not found".into(),
358                    };
359                };
360                self
361                    .invoke(
362                        Some(pending.tool),
363                        pending.parameters,
364                        pending.conversation_id,
365                        pending.identity,
366                        pending.channel,
367                        request.authentication,
368                        Some(challenge_response),
369                    )
370                    .await
371            }
372            _ => QefroResponse::Error {
373                code: "invalid_request".into(),
374                message: "Unsupported request type".into(),
375            },
376        }
377    }
378
379    async fn invoke(
380        &mut self,
381        tool: Option<String>,
382        parameters: Value,
383        conversation_id: Uuid,
384        identity: Option<Value>,
385        channel: Option<String>,
386        authentication: Option<Value>,
387        auth_response: Option<String>,
388    ) -> QefroResponse {
389        let Some(tool_name) = tool else {
390            return QefroResponse::Error {
391                code: "invalid_request".into(),
392                message: "tool is required".into(),
393            };
394        };
395
396        let Some(registration) = self.tools.get(&tool_name).cloned() else {
397            return QefroResponse::Error {
398                code: "not_found".into(),
399                message: format!("Unknown tool: {tool_name}"),
400            };
401        };
402
403        let mut current_customer = self
404            .auth_by_conversation
405            .get(&conversation_id)
406            .filter(|a| a.expires_at_epoch_ms > Utc::now().timestamp_millis())
407            .map(|a| a.customer.clone());
408
409        if registration.metadata.auth == ToolAuthMode::Required {
410            match self
411                .ensure_authorized_customer(
412                    &tool_name,
413                    &registration.metadata,
414                    &parameters,
415                    conversation_id,
416                    identity.clone(),
417                    channel.clone(),
418                    auth_response.clone(),
419                )
420                .await
421            {
422                Ok(customer) => {
423                    current_customer = Some(customer);
424                }
425                Err(resp) => return resp,
426            }
427        }
428
429        let ctx = ToolContext {
430            identity: identity.clone().unwrap_or_else(|| json!({})),
431            parameters: parameters.clone(),
432            conversation_id,
433            channel,
434            authentication,
435            auth_response,
436            customer: current_customer,
437        };
438
439        match (registration.handler)(ctx).await {
440            Ok(output) => {
441                let auth = self
442                    .auth_by_conversation
443                    .get(&conversation_id)
444                    .filter(|v| v.expires_at_epoch_ms > Utc::now().timestamp_millis())
445                    .map(|v| v.auth.clone());
446                QefroResponse::Result {
447                    output,
448                    authentication_context: auth,
449                }
450            }
451            Err(e) => QefroResponse::Error {
452                code: "internal_error".into(),
453                message: e.to_string(),
454            },
455        }
456    }
457
458    async fn ensure_authorized_customer(
459        &mut self,
460        tool: &str,
461        metadata: &ToolMetadata,
462        parameters: &Value,
463        conversation_id: Uuid,
464        identity: Option<Value>,
465        channel: Option<String>,
466        auth_response: Option<String>,
467    ) -> std::result::Result<Value, QefroResponse> {
468        if let Some(existing) = self
469            .auth_by_conversation
470            .get(&conversation_id)
471            .filter(|a| a.expires_at_epoch_ms > Utc::now().timestamp_millis())
472        {
473            return Ok(existing.customer.clone());
474        }
475
476        let Some(provider) = self.customer_provider.as_ref() else {
477            return Err(QefroResponse::Error {
478                code: "configuration_error".into(),
479                message: "Tool requires customer provider. Configure app.customer(...)".into(),
480            });
481        };
482
483        let lookup = CustomerLookupContext {
484            identity: identity.clone().unwrap_or_else(|| json!({})),
485            parameters: parameters.clone(),
486            conversation_id,
487            channel: channel.clone(),
488        };
489
490        let Some(customer) = provider.lookup(&lookup).await.map_err(|e| QefroResponse::Error {
491            code: "internal_error".into(),
492            message: e.to_string(),
493        })? else {
494            return Err(QefroResponse::Error {
495                code: "customer_not_found".into(),
496                message: "Customer not found".into(),
497            });
498        };
499
500        let authorize = CustomerAuthorizeContext {
501            customer,
502            method: metadata.default_auth_method.clone(),
503            response: auth_response,
504            identity: identity.unwrap_or_else(|| json!({})),
505            parameters: parameters.clone(),
506            conversation_id,
507            channel,
508        };
509
510        let outcome = provider
511            .authorize(&authorize)
512            .await
513            .map_err(|e| QefroResponse::Error {
514                code: "internal_error".into(),
515                message: e.to_string(),
516            })?;
517
518        self.require_authentication(
519            conversation_id,
520            outcome,
521            tool,
522            parameters.clone(),
523            Some(authorize.identity),
524            authorize.channel,
525        )
526    }
527
528    pub fn require_authentication(
529        &mut self,
530        conversation_id: Uuid,
531        outcome: AuthOutcome,
532        tool: &str,
533        parameters: Value,
534        identity: Option<Value>,
535        channel: Option<String>,
536    ) -> std::result::Result<Value, QefroResponse> {
537        match outcome {
538            AuthOutcome::Success { customer, auth } => {
539                let ttl = auth.expires_in.unwrap_or(900).max(1);
540                self.auth_by_conversation.insert(
541                    conversation_id,
542                    StoredAuth {
543                        customer: customer.clone(),
544                        auth,
545                        expires_at_epoch_ms: Utc::now().timestamp_millis() + ttl * 1000,
546                    },
547                );
548                Ok(customer)
549            }
550            AuthOutcome::Challenge(challenge) => {
551                let resume_token = Uuid::new_v4().to_string();
552                self.pending.insert(
553                    resume_token.clone(),
554                    PendingInvocation {
555                        tool: tool.to_string(),
556                        conversation_id,
557                        parameters,
558                        identity,
559                        channel,
560                    },
561                );
562                Err(QefroResponse::Challenge {
563                    resume_token,
564                    challenge,
565                })
566            }
567            AuthOutcome::Denied => Err(QefroResponse::Error {
568                code: "denied".into(),
569                message: "Authentication denied".into(),
570            }),
571            AuthOutcome::NotFound => Err(QefroResponse::Error {
572                code: "customer_not_found".into(),
573                message: "Customer not found".into(),
574            }),
575        }
576    }
577}
qefro_backend_sdk/lib.rs

qefro_backend_sdk/
lib.rs
