Vakyartha documentation

892 )
893 })?
894}
895
896/// DELETE /feeds/sources/{name} — Remove a feed source from config.
897/// DELETE /feeds/sources/{name} — Remove a feed source.
898///
899/// Soft-deletes the materialized source row and removes its declaration from
900/// the canonical scope-specific configuration.
901pub async fn delete_feed_source(
902 Path(source_id): Path<String>,
903 Query(params): Query<HashMap<String, String>>,
904 State(state): State<AppState>,
905) -> Result<impl IntoResponse, (StatusCode, String)> {
906 let cwd = state.core.cwd();
907 let scope = params
908 .get("scope")
909 .map(String::as_str)
910 .unwrap_or("workspace");
911 authorize_feed_mutation(&state, scope)?;
912
913 let removed = run_feed_admin_script(
914 cwd,
915 "feed_ingest.py",
916 &["--remove-source", &source_id, "--scope", scope],
917 )
918 .await?
919 .get("status")
920 .and_then(|s| s.as_str())
921 == Some("ok");
922
923 if !removed {
924 return Err((
925 StatusCode::NOT_FOUND,
926 format!("Source '{}' not found", source_id),
927 ));
928 }
929
930 let config_path = if scope == "global" {
931 global_feeds_config_path()
932 } else {
933 feeds_config_path(cwd)
934 };
935 let declared = source_id.clone();
936 let _ = edit_feeds_config(config_path, move |current| {
937 let next =
938 current.and_then(|content| remove_array_table_block(content, "sources", &declared));
939 Ok((next, ()))
940 })
941 .await;
942
943 Ok(Json(json!({
944 "status": "ok",
945 "message": format!("Source '{}' removed", source_id),
946 })))
947}
948
949/// PATCH /feeds/sources/{name} — Update a source's enabled/interval/trust fields.
950///
951/// `tags` is **refused** here. The `feeds` table has no tags column, so a
952/// tag written on update reached feeds.toml and nothing else: the DB never
953/// saw it and the UI never showed it. A PATCH that silently half-applies
954/// is worse than one that says no, so the caller is told where tags are
955/// actually set instead (AGENTS.md invariant 30).
956pub async fn update_feed_source(
957 Path(source_id): Path<String>,
958 Query(params): Query<HashMap<String, String>>,
959 State(state): State<AppState>,
960 Json(payload): Json<Value>,
961) -> Result<impl IntoResponse, (StatusCode, String)> {
962 let cwd = state.core.cwd();
963 let scope = params
964 .get("scope")
965 .map(String::as_str)
966 .unwrap_or("workspace");
967 authorize_feed_mutation(&state, scope)?;
968
969 let enabled = payload.get("enabled").and_then(|v| v.as_bool());
970 let interval = payload.get("interval").and_then(|v| v.as_str());
971 let trust = payload.get("trust").and_then(|v| v.as_str());
972 if payload.get("tags").is_some() {
973 return Err((
974 StatusCode::BAD_REQUEST,
975 "tags cannot be updated here: the feeds table has no tags column, so the \
976 change would reach feeds.toml and nothing else. Set tags when creating \
977 the source."
978 .into(),
979 ));
980 }
981
982 if enabled.is_none() && interval.is_none() && trust.is_none() {
983 return Err((
984 StatusCode::BAD_REQUEST,
985 "No recognized fields to update (enabled, interval, trust)".into(),
986 ));
987 }
988
989 if enabled.is_some() || interval.is_some() || trust.is_some() {
990 let mut args: Vec<String> = vec!["--update-source".into(), source_id.clone()];
991 args.extend(["--scope".into(), scope.into()]);
992 if let Some(e) = enabled {
993 args.push("--set-enabled".into());
994 args.push(if e { "true" } else { "false" }.into());
995 }
996 if let Some(i) = interval {
997 args.push("--set-interval".into());
998 args.push(i.into());
999 }
1000 if let Some(t) = trust {
1001 args.push("--set-trust".into());
1002 args.push(t.into());
1003 }
1004 let arg_refs: Vec<&str> = args.iter().map(|s| s.as_str()).collect();
1005 let updated = run_feed_admin_script(cwd, "feed_ingest.py", &arg_refs)
1006 .await?
1007 .get("status")
1008 .and_then(|s| s.as_str())
1009 == Some("ok");
1010 if !updated {
1011 return Err((
1012 StatusCode::NOT_FOUND,
1013 format!("Source '{}' not found", source_id),
1014 ));
1015 }
1016 }
1017
1018 // Reconcile the durable declaration when this source has one. The
1019 // materialized row above remains authoritative for the current response.
1020 let config_path = if scope == "global" {
1021 global_feeds_config_path()
1022 } else {
1023 feeds_config_path(cwd)
1024 };
1025 let mut fields = Vec::new();
1026 if let Some(e) = enabled {
1027 fields.push(("enabled", format!("enabled = {e}")));
1028 }
1029 if let Some(i) = interval {
1030 fields.push(("interval", format!("interval = \"{}\"", escape_toml(i))));
1031 }
1032 if let Some(t) = trust {
1033 fields.push(("trust", format!("trust = \"{}\"", escape_toml(t))));
1034 }
1035 let declared = source_id.clone();
1036 let _ = edit_feeds_config(config_path, move |current| {
1037 let Some(mut content) = current.map(ToOwned::to_owned) else {
1038 return Ok((None, ()));
1039 };
1040 let mut touched = false;
1041 for (field, line) in &fields {
1042 if let Some(next) = set_field_in_block(&content, "sources", &declared, field, line) {
1043 content = next;
1044 touched = true;
1045 }
1046 }
1047 Ok((touched.then_some(content), ()))
1048 })
1049 .await;
1050
1051 Ok(Json(json!({
1052 "status": "ok",
1053 "message": format!("Source '{}' updated", source_id),
1054 })))
1055}
1056
1057/// GET /feeds/sources/configured — List sources actually known to the DB
1058/// (populated by ingestion), with live enabled/trust/interval status.
1059pub async fn list_configured_sources(
1060 State(state): State<AppState>,
1061) -> Result<impl IntoResponse, (StatusCode, String)> {
1062 let cwd = state.core.cwd();
1063 let request = json!({
1064 "jsonrpc": "2.0",
1065 "id": 1,
1066 "method": "tools/call",
1067 "params": {
1068 "name": "feed_sources",
1069 "arguments": {}
1070 }
1071 });
1072 let result = run_feed_mcp_request(cwd, &request).await?;
1073 Ok(Json(result))
1074}
1075
1076/// POST /feeds/alerts — Add a new alert rule to feeds.toml.
1077pub async fn add_feed_alert(
1078 State(state): State<AppState>,
1079 Json(payload): Json<Value>,
1080) -> Result<impl IntoResponse, (StatusCode, String)> {
1081 let cwd = state.core.cwd();
1082 let scope = payload
1083 .get("scope")
1084 .and_then(Value::as_str)
1085 .unwrap_or("workspace");
1086 authorize_feed_mutation(&state, scope)?;
1087 let config_path = if scope == "global" {
1088 global_feeds_config_path()
1089 } else {
1090 feeds_config_path(cwd)
1091 };
1092
1093 let name = payload
1094 .get("name")
1095 .and_then(|v| v.as_str())
1096 .filter(|s| !s.is_empty())
1097 .ok_or((StatusCode::BAD_REQUEST, "Alert 'name' is required".into()))?;
1098
1099 let str_array = |key: &str| -> Vec<String> {
1100 payload
1101 .get(key)
1102 .and_then(|v| v.as_array())
1103 .map(|a| {
1104 a.iter()
1105 .filter_map(|v| v.as_str())
1106 .map(String::from)
1107 .collect()
1108 })
1109 .unwrap_or_default()
1110 };
1111 let keywords = str_array("keywords");
1112 let tags = str_array("tags");
1113 let sources = str_array("sources");
1114
1115 if keywords.is_empty() && tags.is_empty() && sources.is_empty() {
1116 return Err((
1117 StatusCode::BAD_REQUEST,
1118 "At least one of keywords, tags, or sources must be set".into(),
1119 ));
1120 }
1121
1122 let action = payload
1123 .get("action")
1124 .and_then(|v| v.as_str())
1125 .unwrap_or("deliver");
1126 if !matches!(action, "deliver" | "hook" | "both") {
1127 return Err((
1128 StatusCode::BAD_REQUEST,
1129 "action must be one of: deliver, hook, both".into(),
1130 ));
1131 }
1132 let deliver_to = payload
1133 .get("deliver_to")
1134 .and_then(|v| v.as_str())
1135 .unwrap_or("");
1136 if matches!(action, "deliver" | "both") && deliver_to.trim().is_empty() {
1137 return Err((
1138 StatusCode::BAD_REQUEST,
1139 "deliver_to is required for deliver and both alerts".into(),
1140 ));
1141 }
1142 let cooldown_minutes = payload
1143 .get("cooldown_minutes")
1144 .and_then(|v| v.as_i64())
1145 .unwrap_or(30);
1146
1147 let to_toml_array = |items: &[String]| -> String {
1148 items
1149 .iter()
1150 .map(|t| format!("\"{}\"", escape_toml(t)))
1151 .collect::<Vec<_>>()
1152 .join(", ")
1153 };
1154
1155 let mut block = format!(
1156 "\n[[alerts]]\nname = \"{}\"\naction = \"{}\"\ncooldown_minutes = {}\nenabled = true\n",
1157 escape_toml(name),
1158 escape_toml(action),
1159 cooldown_minutes
1160 );
1161 if !deliver_to.is_empty() {
1162 block.push_str(&format!("deliver_to = \"{}\"\n", escape_toml(deliver_to)));
1163 }
1164 block.push_str("\n[alerts.match]\n");
1165 if !keywords.is_empty() {
1166 block.push_str(&format!("keywords = [{}]\n", to_toml_array(&keywords)));
1167 }
1168 if !tags.is_empty() {
1169 block.push_str(&format!("tags = [{}]\n", to_toml_array(&tags)));
1170 }
1171 if !sources.is_empty() {
1172 block.push_str(&format!("sources = [{}]\n", to_toml_array(&sources)));
1173 }
1174
1175 edit_feeds_config(config_path.clone(), move |current| {
1176 let mut content = current.map_or_else(default_feeds_config, ToOwned::to_owned);
1177 append_block(&mut content, &block);
1178 Ok((Some(content), ()))
1179 })
1180 .await?;
1181 run_feed_script(cwd, "feed_ingest.py", &["--sync-alerts"]).await?;
1182
1183 Ok(Json(json!({
1184 "status": "ok",
1185 "message": format!("Alert '{}' added to {}", name, config_path.display()),
1186 })))
1187}
1188
1189/// DELETE /feeds/alerts/{name} — Remove an alert rule from config.
1190pub async fn delete_feed_alert(
1191 Path(name): Path<String>,
1192 Query(params): Query<HashMap<String, String>>,
1193 State(state): State<AppState>,
1194) -> Result<impl IntoResponse, (StatusCode, String)> {
1195 let cwd = state.core.cwd();
1196 let scope = params
1197 .get("scope")
1198 .map(String::as_str)
1199 .unwrap_or("workspace");
1200 authorize_feed_mutation(&state, scope)?;
1201 let config_path = if scope == "global" {
1202 global_feeds_config_path()
1203 } else {
1204 feeds_config_path(cwd)
1205 };
1206
1207 let alert = name.clone();
1208 edit_feeds_config(config_path.clone(), move |current| {
1209 let content =
1210 current.ok_or_else(|| (StatusCode::NOT_FOUND, "Config file not found".to_string()))?;
1211 let next = remove_array_table_block(content, "alerts", &alert)
1212 .ok_or_else(|| (StatusCode::NOT_FOUND, format!("Alert '{alert}' not found")))?;
1213 Ok((Some(next), ()))
1214 })
1215 .await?;
1216 run_feed_script(cwd, "feed_ingest.py", &["--sync-alerts"]).await?;
1217
1218 Ok(Json(json!({
1219 "status": "ok",
1220 "message": format!("Alert '{}' removed from {}", name, config_path.display()),
1221 })))
1222}
1223
1224/// Build feed routes.
1225pub fn routes() -> Router<AppState> {
1226 Router::new()
1227 .route(
1228 "/feeds/sources",
1229 get(list_source_types).post(add_feed_source),
1230 )
1231 .route("/feeds/sources/configured", get(list_configured_sources))
1232 .route(
1233 "/feeds/sources/{name}",
1234 delete(delete_feed_source).patch(update_feed_source),
1235 )
1236 .route("/feeds/config", get(get_feed_config))
1237 .route("/feeds/items", get(list_feed_items))
1238 .route("/feeds/items/{id}", get(get_feed_item))
1239 .route("/feeds/search", get(search_feed_items))
1240 .route("/feeds/stats", get(get_feed_stats))
1241 .route("/feeds/alerts", get(get_feed_alerts).post(add_feed_alert))
1242 .route("/feeds/runs", get(get_feed_runs))
1243 .route("/feeds/quarantine", get(get_feed_quarantine))
1244 .route("/feeds/quarantine/{id}/release", post(release_feed_item))
1245 .route("/feeds/alerts/{name}", delete(delete_feed_alert))
1246 .route("/feeds/ingest", post(trigger_ingestion))
1247}
1248
1249#[cfg(test)]
1250#[allow(clippy::panic, clippy::unwrap_used, clippy::expect_used)]
1251mod tests {
1252 use super::{authorize_feed_scope, feed_environment, feeds_dir, validate_source_url};
1253 use vak_config::PermissionMode;
1254
1255 /// The feed subprocess runs with `env_clear`, so `feed_environment` is the
1256 /// complete list of what it sees. It must carry the operational minimum
1257 /// (PATH, so `python3` resolves) and never a parent secret — before this,
1258 /// the subprocess inherited the whole server environment, `VAK_GATEWAY_TOKEN`
1259 /// and provider keys included (invariant 12).
1260 #[test]
1261 fn feed_subprocess_environment_is_a_secret_free_allowlist() {
1262 let _home = vak_config::paths::isolate_home_for_tests();
1263 let environment = feed_environment(
1264 std::path::Path::new("/tmp/ws"),
1265 std::path::Path::new("/tmp/scripts"),
1266 );
1267 let allowed = [
1268 "PYTHONPATH",
1269 "VAK_FEED_WORKSPACE",
1270 "VAK_FEEDS_DB",
1271 "VAK_FEEDS_LOG",
1272 "VAK_FEEDS_CONFIG",
1273 "PATH",
1274 "HOME",
1275 ];
1276 for (key, _) in &environment {
1277 assert!(allowed.contains(key), "{key} is not in the feed allowlist");
1278 }
1279 assert!(
1280 environment.iter().any(|(key, _)| *key == "PATH"),
1281 "PATH must be passed so python3 resolves under env_clear"
1282 );
1283 for secret in ["VAK_GATEWAY_TOKEN", "ANTHROPIC_API_KEY", "OPENAI_API_KEY"] {
1284 assert!(
1285 !environment.iter().any(|(key, _)| *key == secret),
1286 "{secret} must never be passed to a feed subprocess"
1287 );
1288 }
1289 }
1290
1291 /// The script the server runs is located only from the binary, never from
1292 /// the session workspace: a workspace with its own `scripts/feeds` cannot
1293 /// supply the code the server executes (invariants 12, 14, 15).
1294 #[test]
1295 fn feeds_dir_never_resolves_from_a_workspace() {
1296 let workspace = tempfile::tempdir().expect("tempdir");
1297 let planted = workspace.path().join("scripts").join("feeds");
1298 std::fs::create_dir_all(&planted).expect("mkdir");
1299 std::fs::write(planted.join("feed_ingest.py"), b"raise SystemExit\n").expect("write");
1300 assert!(
1301 !feeds_dir().starts_with(workspace.path()),
1302 "feeds_dir resolved a script from the session workspace"
1303 );
1304 }
1305
1306 /// Every path a feed script writes comes from the canonical data home,
1307 /// so an overridden `VAK_HOME` holds the feed store, its log and its
1308 /// config; the scripts used to work out the platform default for
1309 /// themselves and wrote outside it.
1310 #[test]
1311 fn feeds_write_under_overridden_home() {
1312 let home = vak_config::paths::isolate_home_for_tests();
1313 let environment = feed_environment(
1314 std::path::Path::new("/tmp/ws"),
1315 std::path::Path::new("/tmp/scripts"),
1316 );
1317 for name in ["VAK_FEEDS_DB", "VAK_FEEDS_LOG", "VAK_FEEDS_CONFIG"] {
1318 let (_, value) = environment
1319 .iter()
1320 .find(|(key, _)| *key == name)
1321 .unwrap_or_else(|| panic!("{name} is passed"));
1322 assert!(
1323 std::path::Path::new(value).starts_with(&home),
1324 "{name} = {value} is outside {}",
1325 home.display()
1326 );
1327 }
1328 }
1329
1330 #[test]
1331 fn feed_scope_permissions_only_tighten() {
1332 assert!(authorize_feed_scope("workspace", PermissionMode::WorkspaceWrite).is_ok());
1333 assert!(authorize_feed_scope("workspace", PermissionMode::FullAccess).is_ok());
1334 assert!(authorize_feed_scope("global", PermissionMode::FullAccess).is_ok());
1335 assert!(authorize_feed_scope("global", PermissionMode::WorkspaceWrite).is_err());
1336 assert!(authorize_feed_scope("workspace", PermissionMode::ReadOnly).is_err());
1337 assert!(authorize_feed_scope("global", PermissionMode::ReadOnly).is_err());
1338 }
1339
1340 #[test]
1341 fn source_admission_rejects_private_and_malformed_targets() {
1342 assert!(validate_source_url("https://example.com/feed").is_ok());
1343 assert!(validate_source_url("not a url").is_err());
1344 assert!(validate_source_url("file:///tmp/feed").is_err());
1345 assert!(validate_source_url("http://127.0.0.1/feed").is_err());
1346 assert!(validate_source_url("http://192.168.1.10/feed").is_err());
1347 assert!(validate_source_url("http://[::1]/feed").is_err());
1348 }
1349}
1350

892 )
893 })?
894}
895
896/// DELETE /feeds/sources/{name} — Remove a feed source from config.
897/// DELETE /feeds/sources/{name} — Remove a feed source.
898///
899/// Soft-deletes the materialized source row and removes its declaration from
900/// the canonical scope-specific configuration.
901pub async fn delete_feed_source(
902 Path(source_id): Path<String>,
903 Query(params): Query<HashMap<String, String>>,
904 State(state): State<AppState>,
905) -> Result<impl IntoResponse, (StatusCode, String)> {
906 let cwd = state.core.cwd();
907 let scope = params
908 .get("scope")
909 .map(String::as_str)
910 .unwrap_or("workspace");
911 authorize_feed_mutation(&state, scope)?;
912
913 let removed = run_feed_admin_script(
914 cwd,
915 "feed_ingest.py",
916 &["--remove-source", &source_id, "--scope", scope],
917 )
918 .await?
919 .get("status")
920 .and_then(|s| s.as_str())
921 == Some("ok");
922
923 if !removed {
924 return Err((
925 StatusCode::NOT_FOUND,
926 format!("Source '{}' not found", source_id),
927 ));
928 }
929
930 let config_path = if scope == "global" {
931 global_feeds_config_path()
932 } else {
933 feeds_config_path(cwd)
934 };
935 let declared = source_id.clone();
936 let _ = edit_feeds_config(config_path, move |current| {
937 let next =
938 current.and_then(|content| remove_array_table_block(content, "sources", &declared));
939 Ok((next, ()))
940 })
941 .await;
942
943 Ok(Json(json!({
944 "status": "ok",
945 "message": format!("Source '{}' removed", source_id),
946 })))
947}
948
949/// PATCH /feeds/sources/{name} — Update a source's enabled/interval/trust fields.
950///
951/// `tags` is **refused** here. The `feeds` table has no tags column, so a
952/// tag written on update reached feeds.toml and nothing else: the DB never
953/// saw it and the UI never showed it. A PATCH that silently half-applies
954/// is worse than one that says no, so the caller is told where tags are
955/// actually set instead (AGENTS.md invariant 30).
956pub async fn update_feed_source(
957 Path(source_id): Path<String>,
958 Query(params): Query<HashMap<String, String>>,
959 State(state): State<AppState>,
960 Json(payload): Json<Value>,
961) -> Result<impl IntoResponse, (StatusCode, String)> {
962 let cwd = state.core.cwd();
963 let scope = params
964 .get("scope")
965 .map(String::as_str)
966 .unwrap_or("workspace");
967 authorize_feed_mutation(&state, scope)?;
968
969 let enabled = payload.get("enabled").and_then(|v| v.as_bool());
970 let interval = payload.get("interval").and_then(|v| v.as_str());
971 let trust = payload.get("trust").and_then(|v| v.as_str());
972 if payload.get("tags").is_some() {
973 return Err((
974 StatusCode::BAD_REQUEST,
975 "tags cannot be updated here: the feeds table has no tags column, so the \
976 change would reach feeds.toml and nothing else. Set tags when creating \
977 the source."
978 .into(),
979 ));
980 }
981
982 if enabled.is_none() && interval.is_none() && trust.is_none() {
983 return Err((
984 StatusCode::BAD_REQUEST,
985 "No recognized fields to update (enabled, interval, trust)".into(),
986 ));
987 }
988
989 if enabled.is_some() || interval.is_some() || trust.is_some() {
990 let mut args: Vec<String> = vec!["--update-source".into(), source_id.clone()];
991 args.extend(["--scope".into(), scope.into()]);
992 if let Some(e) = enabled {
993 args.push("--set-enabled".into());
994 args.push(if e { "true" } else { "false" }.into());
995 }
996 if let Some(i) = interval {
997 args.push("--set-interval".into());
998 args.push(i.into());
999 }
1000 if let Some(t) = trust {
1001 args.push("--set-trust".into());
1002 args.push(t.into());
1003 }
1004 let arg_refs: Vec<&str> = args.iter().map(|s| s.as_str()).collect();
1005 let updated = run_feed_admin_script(cwd, "feed_ingest.py", &arg_refs)
1006 .await?
1007 .get("status")
1008 .and_then(|s| s.as_str())
1009 == Some("ok");
1010 if !updated {
1011 return Err((
1012 StatusCode::NOT_FOUND,
1013 format!("Source '{}' not found", source_id),
1014 ));
1015 }
1016 }
1017
1018 // Reconcile the durable declaration when this source has one. The
1019 // materialized row above remains authoritative for the current response.
1020 let config_path = if scope == "global" {
1021 global_feeds_config_path()
1022 } else {
1023 feeds_config_path(cwd)
1024 };
1025 let mut fields = Vec::new();
1026 if let Some(e) = enabled {
1027 fields.push(("enabled", format!("enabled = {e}")));
1028 }
1029 if let Some(i) = interval {
1030 fields.push(("interval", format!("interval = \"{}\"", escape_toml(i))));
1031 }
1032 if let Some(t) = trust {
1033 fields.push(("trust", format!("trust = \"{}\"", escape_toml(t))));
1034 }
1035 let declared = source_id.clone();
1036 let _ = edit_feeds_config(config_path, move |current| {
1037 let Some(mut content) = current.map(ToOwned::to_owned) else {
1038 return Ok((None, ()));
1039 };
1040 let mut touched = false;
1041 for (field, line) in &fields {
1042 if let Some(next) = set_field_in_block(&content, "sources", &declared, field, line) {
1043 content = next;
1044 touched = true;
1045 }
1046 }
1047 Ok((touched.then_some(content), ()))
1048 })
1049 .await;
1050
1051 Ok(Json(json!({
1052 "status": "ok",
1053 "message": format!("Source '{}' updated", source_id),
1054 })))
1055}
1056
1057/// GET /feeds/sources/configured — List sources actually known to the DB
1058/// (populated by ingestion), with live enabled/trust/interval status.
1059pub async fn list_configured_sources(
1060 State(state): State<AppState>,
1061) -> Result<impl IntoResponse, (StatusCode, String)> {
1062 let cwd = state.core.cwd();
1063 let request = json!({
1064 "jsonrpc": "2.0",
1065 "id": 1,
1066 "method": "tools/call",
1067 "params": {
1068 "name": "feed_sources",
1069 "arguments": {}
1070 }
1071 });
1072 let result = run_feed_mcp_request(cwd, &request).await?;
1073 Ok(Json(result))
1074}
1075
1076/// POST /feeds/alerts — Add a new alert rule to feeds.toml.
1077pub async fn add_feed_alert(
1078 State(state): State<AppState>,
1079 Json(payload): Json<Value>,
1080) -> Result<impl IntoResponse, (StatusCode, String)> {
1081 let cwd = state.core.cwd();
1082 let scope = payload
1083 .get("scope")
1084 .and_then(Value::as_str)
1085 .unwrap_or("workspace");
1086 authorize_feed_mutation(&state, scope)?;
1087 let config_path = if scope == "global" {
1088 global_feeds_config_path()
1089 } else {
1090 feeds_config_path(cwd)
1091 };
1092
1093 let name = payload
1094 .get("name")
1095 .and_then(|v| v.as_str())
1096 .filter(|s| !s.is_empty())
1097 .ok_or((StatusCode::BAD_REQUEST, "Alert 'name' is required".into()))?;
1098
1099 let str_array = |key: &str| -> Vec<String> {
1100 payload
1101 .get(key)
1102 .and_then(|v| v.as_array())
1103 .map(|a| {
1104 a.iter()
1105 .filter_map(|v| v.as_str())
1106 .map(String::from)
1107 .collect()
1108 })
1109 .unwrap_or_default()
1110 };
1111 let keywords = str_array("keywords");
1112 let tags = str_array("tags");
1113 let sources = str_array("sources");
1114
1115 if keywords.is_empty() && tags.is_empty() && sources.is_empty() {
1116 return Err((
1117 StatusCode::BAD_REQUEST,
1118 "At least one of keywords, tags, or sources must be set".into(),
1119 ));
1120 }
1121
1122 let action = payload
1123 .get("action")
1124 .and_then(|v| v.as_str())
1125 .unwrap_or("deliver");
1126 if !matches!(action, "deliver" | "hook" | "both") {
1127 return Err((
1128 StatusCode::BAD_REQUEST,
1129 "action must be one of: deliver, hook, both".into(),
1130 ));
1131 }
1132 let deliver_to = payload
1133 .get("deliver_to")
1134 .and_then(|v| v.as_str())
1135 .unwrap_or("");
1136 if matches!(action, "deliver" | "both") && deliver_to.trim().is_empty() {
1137 return Err((
1138 StatusCode::BAD_REQUEST,
1139 "deliver_to is required for deliver and both alerts".into(),
1140 ));
1141 }
1142 let cooldown_minutes = payload
1143 .get("cooldown_minutes")
1144 .and_then(|v| v.as_i64())
1145 .unwrap_or(30);
1146
1147 let to_toml_array = |items: &[String]| -> String {
1148 items
1149 .iter()
1150 .map(|t| format!("\"{}\"", escape_toml(t)))
1151 .collect::<Vec<_>>()
1152 .join(", ")
1153 };
1154
1155 let mut block = format!(
1156 "\n[[alerts]]\nname = \"{}\"\naction = \"{}\"\ncooldown_minutes = {}\nenabled = true\n",
1157 escape_toml(name),
1158 escape_toml(action),
1159 cooldown_minutes
1160 );
1161 if !deliver_to.is_empty() {
1162 block.push_str(&format!("deliver_to = \"{}\"\n", escape_toml(deliver_to)));
1163 }
1164 block.push_str("\n[alerts.match]\n");
1165 if !keywords.is_empty() {
1166 block.push_str(&format!("keywords = [{}]\n", to_toml_array(&keywords)));
1167 }
1168 if !tags.is_empty() {
1169 block.push_str(&format!("tags = [{}]\n", to_toml_array(&tags)));
1170 }
1171 if !sources.is_empty() {
1172 block.push_str(&format!("sources = [{}]\n", to_toml_array(&sources)));
1173 }
1174
1175 edit_feeds_config(config_path.clone(), move |current| {
1176 let mut content = current.map_or_else(default_feeds_config, ToOwned::to_owned);
1177 append_block(&mut content, &block);
1178 Ok((Some(content), ()))
1179 })
1180 .await?;
1181 run_feed_script(cwd, "feed_ingest.py", &["--sync-alerts"]).await?;
1182
1183 Ok(Json(json!({
1184 "status": "ok",
1185 "message": format!("Alert '{}' added to {}", name, config_path.display()),
1186 })))
1187}
1188
1189/// DELETE /feeds/alerts/{name} — Remove an alert rule from config.
1190pub async fn delete_feed_alert(
1191 Path(name): Path<String>,
1192 Query(params): Query<HashMap<String, String>>,
1193 State(state): State<AppState>,
1194) -> Result<impl IntoResponse, (StatusCode, String)> {
1195 let cwd = state.core.cwd();
1196 let scope = params
1197 .get("scope")
1198 .map(String::as_str)
1199 .unwrap_or("workspace");
1200 authorize_feed_mutation(&state, scope)?;
1201 let config_path = if scope == "global" {
1202 global_feeds_config_path()
1203 } else {
1204 feeds_config_path(cwd)
1205 };
1206
1207 let alert = name.clone();
1208 edit_feeds_config(config_path.clone(), move |current| {
1209 let content =
1210 current.ok_or_else(|| (StatusCode::NOT_FOUND, "Config file not found".to_string()))?;
1211 let next = remove_array_table_block(content, "alerts", &alert)
1212 .ok_or_else(|| (StatusCode::NOT_FOUND, format!("Alert '{alert}' not found")))?;
1213 Ok((Some(next), ()))
1214 })
1215 .await?;
1216 run_feed_script(cwd, "feed_ingest.py", &["--sync-alerts"]).await?;
1217
1218 Ok(Json(json!({
1219 "status": "ok",
1220 "message": format!("Alert '{}' removed from {}", name, config_path.display()),
1221 })))
1222}
1223
1224/// Build feed routes.
1225pub fn routes() -> Router<AppState> {
1226 Router::new()
1227 .route(
1228 "/feeds/sources",
1229 get(list_source_types).post(add_feed_source),
1230 )
1231 .route("/feeds/sources/configured", get(list_configured_sources))
1232 .route(
1233 "/feeds/sources/{name}",
1234 delete(delete_feed_source).patch(update_feed_source),
1235 )
1236 .route("/feeds/config", get(get_feed_config))
1237 .route("/feeds/items", get(list_feed_items))
1238 .route("/feeds/items/{id}", get(get_feed_item))
1239 .route("/feeds/search", get(search_feed_items))
1240 .route("/feeds/stats", get(get_feed_stats))
1241 .route("/feeds/alerts", get(get_feed_alerts).post(add_feed_alert))
1242 .route("/feeds/runs", get(get_feed_runs))
1243 .route("/feeds/quarantine", get(get_feed_quarantine))
1244 .route("/feeds/quarantine/{id}/release", post(release_feed_item))
1245 .route("/feeds/alerts/{name}", delete(delete_feed_alert))
1246 .route("/feeds/ingest", post(trigger_ingestion))
1247}
1248
1249#[cfg(test)]
1250#[allow(clippy::panic, clippy::unwrap_used, clippy::expect_used)]
1251mod tests {
1252 use super::{authorize_feed_scope, feed_environment, feeds_dir, validate_source_url};
1253 use vak_config::PermissionMode;
1254
1255 /// The feed subprocess runs with `env_clear`, so `feed_environment` is the
1256 /// complete list of what it sees. It must carry the operational minimum
1257 /// (PATH, so `python3` resolves) and never a parent secret — before this,
1258 /// the subprocess inherited the whole server environment, `VAK_GATEWAY_TOKEN`
1259 /// and provider keys included (invariant 12).
1260 #[test]
1261 fn feed_subprocess_environment_is_a_secret_free_allowlist() {
1262 let _home = vak_config::paths::isolate_home_for_tests();
1263 let environment = feed_environment(
1264 std::path::Path::new("/tmp/ws"),
1265 std::path::Path::new("/tmp/scripts"),
1266 );
1267 let allowed = [
1268 "PYTHONPATH",
1269 "VAK_FEED_WORKSPACE",
1270 "VAK_FEEDS_DB",
1271 "VAK_FEEDS_LOG",
1272 "VAK_FEEDS_CONFIG",
1273 "PATH",
1274 "HOME",
1275 ];
1276 for (key, _) in &environment {
1277 assert!(allowed.contains(key), "{key} is not in the feed allowlist");
1278 }
1279 assert!(
1280 environment.iter().any(|(key, _)| *key == "PATH"),
1281 "PATH must be passed so python3 resolves under env_clear"
1282 );
1283 for secret in ["VAK_GATEWAY_TOKEN", "ANTHROPIC_API_KEY", "OPENAI_API_KEY"] {
1284 assert!(
1285 !environment.iter().any(|(key, _)| *key == secret),
1286 "{secret} must never be passed to a feed subprocess"
1287 );
1288 }
1289 }
1290
1291 /// The script the server runs is located only from the binary, never from
1292 /// the session workspace: a workspace with its own `scripts/feeds` cannot
1293 /// supply the code the server executes (invariants 12, 14, 15).
1294 #[test]
1295 fn feeds_dir_never_resolves_from_a_workspace() {
1296 let workspace = tempfile::tempdir().expect("tempdir");
1297 let planted = workspace.path().join("scripts").join("feeds");
1298 std::fs::create_dir_all(&planted).expect("mkdir");
1299 std::fs::write(planted.join("feed_ingest.py"), b"raise SystemExit\n").expect("write");
1300 assert!(
1301 !feeds_dir().starts_with(workspace.path()),
1302 "feeds_dir resolved a script from the session workspace"
1303 );
1304 }
1305
1306 /// Every path a feed script writes comes from the canonical data home,
1307 /// so an overridden `VAK_HOME` holds the feed store, its log and its
1308 /// config; the scripts used to work out the platform default for
1309 /// themselves and wrote outside it.
1310 #[test]
1311 fn feeds_write_under_overridden_home() {
1312 let home = vak_config::paths::isolate_home_for_tests();
1313 let environment = feed_environment(
1314 std::path::Path::new("/tmp/ws"),
1315 std::path::Path::new("/tmp/scripts"),
1316 );
1317 for name in ["VAK_FEEDS_DB", "VAK_FEEDS_LOG", "VAK_FEEDS_CONFIG"] {
1318 let (_, value) = environment
1319 .iter()
1320 .find(|(key, _)| *key == name)
1321 .unwrap_or_else(|| panic!("{name} is passed"));
1322 assert!(
1323 std::path::Path::new(value).starts_with(&home),
1324 "{name} = {value} is outside {}",
1325 home.display()
1326 );
1327 }
1328 }
1329
1330 #[test]
1331 fn feed_scope_permissions_only_tighten() {
1332 assert!(authorize_feed_scope("workspace", PermissionMode::WorkspaceWrite).is_ok());
1333 assert!(authorize_feed_scope("workspace", PermissionMode::FullAccess).is_ok());
1334 assert!(authorize_feed_scope("global", PermissionMode::FullAccess).is_ok());
1335 assert!(authorize_feed_scope("global", PermissionMode::WorkspaceWrite).is_err());
1336 assert!(authorize_feed_scope("workspace", PermissionMode::ReadOnly).is_err());
1337 assert!(authorize_feed_scope("global", PermissionMode::ReadOnly).is_err());
1338 }
1339
1340 #[test]
1341 fn source_admission_rejects_private_and_malformed_targets() {
1342 assert!(validate_source_url("https://example.com/feed").is_ok());
1343 assert!(validate_source_url("not a url").is_err());
1344 assert!(validate_source_url("file:///tmp/feed").is_err());
1345 assert!(validate_source_url("http://127.0.0.1/feed").is_err());
1346 assert!(validate_source_url("http://192.168.1.10/feed").is_err());
1347 assert!(validate_source_url("http://[::1]/feed").is_err());
1348 }
1349}
1350

Feeds

Indexing the workspace…

Feeds