1use serde_json::{Map, Value};
20use tiberius::ToSql;
21
22use crate::auth::auth_manager::AuthManager;
23use crate::core::config_schema::Config;
24use crate::data::store::EndpointRecord;
25use crate::services::sql_pool;
26use crate::services::sql_type::{column_data_to_json, json_to_param};
27use crate::validation::validator::resolved_schemas_for;
28
29fn parse_path(path: &str) -> anyhow::Result<(&str, &str, &str)> {
35 let mut parts = path.trim_start_matches('/').splitn(3, '/');
36 match (parts.next(), parts.next(), parts.next()) {
37 (Some(db), Some(schema), Some(name))
38 if !db.is_empty() && !schema.is_empty() && !name.is_empty() =>
39 {
40 Ok((
41 validate_ident(db)?,
42 validate_ident(schema)?,
43 validate_ident(name)?,
44 ))
45 }
46 _ => anyhow::bail!(
47 "endpoint path '{path}' is not in the expected /<db>/<schema>/<name> shape"
48 ),
49 }
50}
51
52fn validate_ident(ident: &str) -> anyhow::Result<&str> {
67 if !ident.is_empty()
68 && ident
69 .bytes()
70 .all(|b| b.is_ascii_alphanumeric() || b == b'_')
71 {
72 Ok(ident)
73 } else {
74 anyhow::bail!("identifier '{ident}' contains characters outside [A-Za-z0-9_]")
75 }
76}
77
78fn quote_ident(ident: &str) -> String {
84 format!("[{}]", ident.replace(']', "]]"))
85}
86
87struct Param {
88 name: String,
89 ordinal: u64,
90 x_sql_type: String,
91}
92
93fn request_properties(input_schema: &Value) -> Option<&Map<String, Value>> {
103 let body_schema = input_schema.get("properties")?.get("body")?;
104 let resolved = match body_schema.get("$ref").and_then(Value::as_str) {
105 Some(reference) => reference
106 .strip_prefix("#/$defs/")
107 .and_then(|name| input_schema.get("$defs")?.get(name))?,
108 None => body_schema,
109 };
110 resolved.get("properties").and_then(Value::as_object)
111}
112
113fn ordered_params(input_schema: &Value) -> Vec<Param> {
119 let mut params: Vec<Param> = request_properties(input_schema)
120 .into_iter()
121 .flatten()
122 .map(|(name, schema)| Param {
123 name: name.clone(),
124 ordinal: schema
125 .get("x-sql-ordinal")
126 .and_then(Value::as_u64)
127 .unwrap_or(0),
128 x_sql_type: schema
129 .get("x-sql-type")
130 .and_then(Value::as_str)
131 .unwrap_or("nvarchar(max)")
132 .to_string(),
133 })
134 .collect();
135 params.sort_by_key(|p| p.ordinal);
136 params
137}
138
139fn build_statement(
142 db: &str,
143 schema: &str,
144 name: &str,
145 kind: Option<&str>,
146 params: &[Param],
147 body: &Map<String, Value>,
148 database_override: Option<&str>,
149) -> anyhow::Result<(String, Vec<Box<dyn ToSql>>)> {
150 let qualified = match (db, database_override) {
168 ("sandbox", Some(requested_db)) => format!(
169 "{}.{}.{}",
170 quote_ident(requested_db),
171 quote_ident(schema),
172 quote_ident(name)
173 ),
174 ("sandbox", None) => format!("{}.{}", quote_ident(schema), quote_ident(name)),
175 (_, _) => format!(
176 "{}.{}.{}",
177 quote_ident(db),
178 quote_ident(schema),
179 quote_ident(name)
180 ),
181 };
182
183 let mut bound: Vec<Box<dyn ToSql>> = Vec::with_capacity(params.len());
184 for param in params {
185 let value = body.get(¶m.name).cloned().unwrap_or(Value::Null);
186 bound.push(json_to_param(&value, ¶m.x_sql_type)?);
187 }
188
189 let sql = match kind {
190 Some("VIEW") => format!("SELECT * FROM {qualified}"),
191 Some(k) if k.ends_with("_FUNCTION") || k.contains("TABLE_VALUED_FUNCTION") => {
192 let placeholders = (1..=bound.len())
193 .map(|i| format!("@P{i}"))
194 .collect::<Vec<_>>()
195 .join(", ");
196 format!("SELECT * FROM {qualified}({placeholders})")
197 }
198 _ => {
205 if bound.is_empty() {
206 format!("EXEC {qualified}")
207 } else {
208 let mut assignments = Vec::with_capacity(params.len());
209 for (i, p) in params.iter().enumerate() {
210 assignments.push(format!("@{} = @P{}", validate_ident(&p.name)?, i + 1));
220 }
221 format!("EXEC {qualified} {}", assignments.join(", "))
222 }
223 }
224 };
225
226 Ok((sql, bound))
227}
228
229fn classify_tiberius_error(err: tiberius::error::Error) -> anyhow::Error {
239 use tiberius::error::Error as TError;
240 match &err {
241 TError::Server(token) => {
242 let (number, state, class, message, procedure, line) = (
243 token.code(),
244 token.state(),
245 token.class(),
246 token.message(),
247 token.procedure(),
248 token.line(),
249 );
250 let category = if class >= 17 {
251 "fatal/resource error (500-equivalent)"
252 } else if class == 14 {
253 "permission denied (403-equivalent)"
254 } else {
255 "statement/user error (400-equivalent)"
256 };
257 anyhow::anyhow!(
258 "SQL Server error {number} (severity {class}, state {state}) in '{procedure}' line {line}: {message} [{category}]"
259 )
260 }
261 other => anyhow::anyhow!("SQL Server connection/protocol error (500-equivalent): {other}"),
262 }
263}
264
265pub struct ApiClient {
266 config: Config,
267}
268
269impl ApiClient {
270 pub fn new(config: Config) -> Self {
271 Self { config }
272 }
273
274 pub async fn execute(
292 &self,
293 endpoint: &EndpointRecord,
294 args: &Value,
295 auth_manager: &mut AuthManager,
296 ) -> anyhow::Result<Value> {
297 let (db, schema, name) = parse_path(&endpoint.path)?;
298
299 let (input_schema, _output_schema) =
300 resolved_schemas_for(&self.config.api_version, &endpoint.operation_id).ok_or_else(
301 || {
302 anyhow::anyhow!(
303 "no resolved schema found for operation '{}' under api_version '{}'",
304 endpoint.operation_id,
305 self.config.api_version
306 )
307 },
308 )?;
309 let params = ordered_params(input_schema);
310
311 let empty = Map::new();
312 let args_map = args.as_object().unwrap_or(&empty);
313 let body = args_map
314 .get("body")
315 .and_then(Value::as_object)
316 .cloned()
317 .unwrap_or_default();
318 let database_override = args_map
319 .get("database")
320 .and_then(Value::as_str)
321 .map(validate_ident)
322 .transpose()?;
323
324 let (sql, bound) = build_statement(
325 db,
326 schema,
327 name,
328 endpoint.description.as_deref(),
329 ¶ms,
330 &body,
331 database_override,
332 )?;
333
334 let auth_method = auth_manager.resolve_tds_auth().await?;
335
336 let (host, port) = self.config.host_and_port();
337 let mut tiberius_config = tiberius::Config::new();
338 tiberius_config.host(host);
339 tiberius_config.port(port);
340 tiberius_config.authentication(auth_method);
341 if self.config.trust_server_cert {
342 tiberius_config.trust_cert();
350 }
351
352 let pool_key = format!("{host}:{port}");
353 let pool =
354 sql_pool::cached_pool(&pool_key, tiberius_config, self.config.pool_max_size).await?;
355 let mut conn = pool.get().await.map_err(|err| {
356 anyhow::anyhow!("failed to obtain a pooled SQL Server connection: {err}")
357 })?;
358
359 let bound_refs: Vec<&dyn ToSql> = bound.iter().map(|param| param.as_ref()).collect();
360 let stream = conn
361 .query(&sql, &bound_refs)
362 .await
363 .map_err(classify_tiberius_error)?;
364 let rows = stream
365 .into_first_result()
366 .await
367 .map_err(classify_tiberius_error)?;
368
369 let json_rows: Vec<Value> = rows
370 .iter()
371 .map(|row| {
372 let mut obj = Map::new();
373 for (column, data) in row.cells() {
374 obj.insert(column.name().to_string(), column_data_to_json(data));
375 }
376 Value::Object(obj)
377 })
378 .collect();
379 Ok(Value::Array(json_rows))
380 }
381}
382
383#[cfg(test)]
384mod tests {
385 use super::*;
386
387 #[test]
388 fn parse_path_splits_db_schema_name() {
389 assert_eq!(
390 parse_path("/master/INFORMATION_SCHEMA/COLUMNS").unwrap(),
391 ("master", "INFORMATION_SCHEMA", "COLUMNS")
392 );
393 }
394
395 #[test]
396 fn parse_path_rejects_a_path_with_too_few_segments() {
397 assert!(parse_path("/master/COLUMNS").is_err());
398 }
399
400 #[test]
401 fn quote_ident_doubles_embedded_closing_brackets() {
402 assert_eq!(quote_ident("weird]name"), "[weird]]name]");
403 }
404
405 #[test]
406 fn validate_ident_accepts_alphanumeric_and_underscore() {
407 assert!(validate_ident("sp_who2").is_ok());
408 assert!(validate_ident("INFORMATION_SCHEMA").is_ok());
409 }
410
411 #[test]
412 fn validate_ident_rejects_anything_else() {
413 assert!(validate_ident("").is_err());
414 assert!(validate_ident("robert'; drop table endpoints;--").is_err());
415 assert!(validate_ident("weird]name").is_err());
416 assert!(validate_ident("has space").is_err());
417 }
418
419 #[test]
420 fn parse_path_rejects_a_path_with_an_unsafe_segment() {
421 assert!(parse_path("/master/sys/sp_who; DROP TABLE endpoints--").is_err());
422 }
423
424 #[test]
425 fn build_statement_selects_from_a_view_with_no_parameters() {
426 let (sql, bound) = build_statement(
427 "master",
428 "INFORMATION_SCHEMA",
429 "COLUMNS",
430 Some("VIEW"),
431 &[],
432 &Map::new(),
433 None,
434 )
435 .unwrap();
436 assert_eq!(sql, "SELECT * FROM [master].[INFORMATION_SCHEMA].[COLUMNS]");
437 assert!(bound.is_empty());
438 }
439
440 #[test]
441 fn build_statement_execs_a_proc_with_named_parameters() {
442 let params = vec![
443 Param {
444 name: "objname".to_string(),
445 ordinal: 1,
446 x_sql_type: "nvarchar(1035)".to_string(),
447 },
448 Param {
449 name: "newname".to_string(),
450 ordinal: 2,
451 x_sql_type: "sysname".to_string(),
452 },
453 ];
454 let mut body = Map::new();
455 body.insert("objname".to_string(), Value::String("t1".to_string()));
456 body.insert("newname".to_string(), Value::String("t2".to_string()));
457 let (sql, bound) = build_statement(
458 "master",
459 "sys",
460 "sp_rename",
461 Some("SQL_STORED_PROCEDURE"),
462 ¶ms,
463 &body,
464 None,
465 )
466 .unwrap();
467 assert_eq!(
468 sql,
469 "EXEC [master].[sys].[sp_rename] @objname = @P1, @newname = @P2"
470 );
471 assert_eq!(bound.len(), 2);
472 }
473
474 #[test]
485 fn build_statement_uses_at_prefixed_names_not_bracket_quoting_for_named_exec_arguments() {
486 let params = vec![Param {
487 name: "table_name".to_string(),
488 ordinal: 1,
489 x_sql_type: "nvarchar(384)".to_string(),
490 }];
491 let mut body = Map::new();
492 body.insert(
493 "table_name".to_string(),
494 Value::String("databases".to_string()),
495 );
496 let (sql, _bound) = build_statement(
497 "sandbox",
498 "sys",
499 "sp_columns",
500 Some("SQL_STORED_PROCEDURE"),
501 ¶ms,
502 &body,
503 None,
504 )
505 .unwrap();
506 assert_eq!(sql, "EXEC [sys].[sp_columns] @table_name = @P1");
507 assert!(!sql.contains("[table_name]"));
508 }
509
510 #[test]
511 fn build_statement_selects_from_a_function_with_positional_placeholders() {
512 let params = vec![Param {
513 name: "session_id".to_string(),
514 ordinal: 1,
515 x_sql_type: "smallint".to_string(),
516 }];
517 let mut body = Map::new();
518 body.insert("session_id".to_string(), Value::from(52));
519 let (sql, bound) = build_statement(
520 "master",
521 "sys",
522 "dm_exec_sql_text",
523 Some("SQL_INLINE_TABLE_VALUED_FUNCTION"),
524 ¶ms,
525 &body,
526 None,
527 )
528 .unwrap();
529 assert_eq!(sql, "SELECT * FROM [master].[sys].[dm_exec_sql_text](@P1)");
530 assert_eq!(bound.len(), 1);
531 }
532
533 #[test]
534 fn build_statement_omits_the_database_qualifier_for_sandbox_by_default() {
535 let (sql, bound) = build_statement(
536 "sandbox",
537 "dbo",
538 "widgets",
539 Some("VIEW"),
540 &[],
541 &Map::new(),
542 None,
543 )
544 .unwrap();
545 assert_eq!(sql, "SELECT * FROM [dbo].[widgets]");
546 assert!(bound.is_empty());
547 }
548
549 #[test]
550 fn build_statement_replaces_sandbox_with_a_requested_database_override() {
551 let (sql, bound) = build_statement(
552 "sandbox",
553 "dbo",
554 "widgets",
555 Some("VIEW"),
556 &[],
557 &Map::new(),
558 Some("reporting"),
559 )
560 .unwrap();
561 assert_eq!(sql, "SELECT * FROM [reporting].[dbo].[widgets]");
562 assert!(bound.is_empty());
563 }
564
565 #[test]
566 fn build_statement_ignores_the_database_override_for_master_and_msdb() {
567 let (sql, _bound) = build_statement(
568 "master",
569 "sys",
570 "sp_who",
571 Some("SQL_STORED_PROCEDURE"),
572 &[],
573 &Map::new(),
574 Some("reporting"),
575 )
576 .unwrap();
577 assert_eq!(sql, "EXEC [master].[sys].[sp_who]");
578 }
579
580 #[test]
581 fn build_statement_execs_a_parameterless_proc() {
582 let (sql, bound) = build_statement(
583 "master",
584 "sys",
585 "sp_who",
586 Some("SQL_STORED_PROCEDURE"),
587 &[],
588 &Map::new(),
589 None,
590 )
591 .unwrap();
592 assert_eq!(sql, "EXEC [master].[sys].[sp_who]");
593 assert!(bound.is_empty());
594 }
595
596 #[test]
597 fn ordered_params_follow_the_schema_ordinals() {
598 let schema = serde_json::json!({
603 "properties": {
604 "body": { "$ref": "#/$defs/sys_example_Request" }
605 },
606 "$defs": {
607 "sys_example_Request": {
608 "properties": {
609 "second": { "x-sql-ordinal": 2, "x-sql-type": "int" },
610 "first": { "x-sql-ordinal": 1, "x-sql-type": "nvarchar(20)" }
611 }
612 }
613 }
614 });
615
616 let params = ordered_params(&schema);
617 assert_eq!(params.len(), 2);
618 assert_eq!(params[0].name, "first");
619 assert_eq!(params[0].x_sql_type, "nvarchar(20)");
620 assert_eq!(params[1].name, "second");
621 }
622
623 #[test]
631 fn ordered_params_resolves_the_body_ref_wrapper() {
632 let schema = serde_json::json!({
633 "properties": {
634 "body": { "$ref": "#/$defs/sys_sp_columns_Request" }
635 },
636 "$defs": {
637 "sys_sp_columns_Request": {
638 "properties": {
639 "table_name": { "x-sql-ordinal": 1, "x-sql-type": "nvarchar(384)" },
640 "table_owner": { "x-sql-ordinal": 2, "x-sql-type": "nvarchar(384)" }
641 }
642 }
643 }
644 });
645
646 let params = ordered_params(&schema);
647 assert_eq!(params.len(), 2);
648 assert_eq!(params[0].name, "table_name");
649 assert_eq!(params[1].name, "table_owner");
650 assert!(params.iter().all(|p| p.name != "body"));
651 }
652
653 #[test]
654 fn ordered_params_resolves_an_inline_body_without_a_ref() {
655 let schema = serde_json::json!({
656 "properties": {
657 "body": {
658 "properties": {
659 "only": { "x-sql-ordinal": 1, "x-sql-type": "int" }
660 }
661 }
662 }
663 });
664
665 let params = ordered_params(&schema);
666 assert_eq!(params.len(), 1);
667 assert_eq!(params[0].name, "only");
668 }
669
670 #[test]
671 fn ordered_params_is_empty_for_a_schema_with_no_body_property() {
672 let schema = serde_json::json!({ "properties": {} });
673 assert!(ordered_params(&schema).is_empty());
674 }
675
676 #[test]
677 fn build_statement_rejects_an_unsafe_parameter_name() {
678 let params = vec![Param {
679 name: "unsafe name".to_string(),
680 ordinal: 1,
681 x_sql_type: "int".to_string(),
682 }];
683
684 assert!(
685 build_statement(
686 "master",
687 "sys",
688 "sp_example",
689 Some("SQL_STORED_PROCEDURE"),
690 ¶ms,
691 &Map::new(),
692 None,
693 )
694 .is_err()
695 );
696 }
697
698 #[test]
699 fn protocol_errors_are_classified_as_server_failures() {
700 let error =
701 classify_tiberius_error(tiberius::error::Error::Protocol("invalid packet".into()));
702 assert_eq!(
703 error.to_string(),
704 "SQL Server connection/protocol error (500-equivalent): Protocol error: invalid packet"
705 );
706 }
707}