fix(passthrough): enforce the body model's rate limit on the raw tunnel

crates/aisix-proxy/src/lib.rs

@@ -2565,6 +2565,319 @@ data: [DONE]\n\n"
         assert_eq!(v["error"]["type"], "rate_limit_exceeded");
     }
 
+    // ---- regression coverage for api7/AISIX-Cloud#1116 --------------
+    // Pre-fix the passthrough tunnel passed `None` to quota::enforce, so
+    // a Model's inline rate_limit (and model-scope policies) never
+    // applied to passthrough traffic — for provider endpoints with no
+    // typed surface (e.g. video generation) the model limit was
+    // unenforceable everywhere. Post-fix the top-level `model` field of
+    // a JSON passthrough body is matched against the addressed
+    // provider's configured Models and its limits reserved like the
+    // typed endpoints.
+
+    fn model_entry_with_rate_limit(
+        name: &str,
+        rate_limit: serde_json::Value,
+    ) -> ResourceEntry<Model> {
+        model_entry_named(name, "gpt-4o", rate_limit)
+    }
+
+    fn model_entry_named(
+        display_name: &str,
+        model_name: &str,
+        rate_limit: serde_json::Value,
+    ) -> ResourceEntry<Model> {
+        let cfg = format!(
+            r#"{{
+                "display_name": "{display_name}",
+                "provider": "openai",
+                "model_name": "{model_name}",
+                "provider_key_id": "{PK_ID}",
+                "rate_limit": {rate_limit}
+            }}"#
+        );
+        let model: Model = serde_json::from_str(&cfg).unwrap();
+        ResourceEntry::new("model-id-1", model, 1)
+    }
+
+    #[tokio::test]
+    async fn passthrough_enforces_model_rate_limit_from_body_model_field() {
+        let upstream = MockServer::start().await;
+        Mock::given(method("POST"))
+            .and(path(
+                "/api/v1/services/aigc/video-generation/video-synthesis",
+            ))
+            .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
+                "output": {"task_id": "t-1", "task_status": "PENDING"}
+            })))
+            .mount(&upstream)
+            .await;
+
+        let hub = Arc::new(Hub::new());
+        hub.register_specialized("openai", Arc::new(openai_test_bridge()));
+        let snap = new_snap(&upstream.uri());
+        snap.models.insert(model_entry_with_rate_limit(
+            "my-video-model",
+            serde_json::json!({"rpm": 1}),
+        ));
+        snap.apikeys
+            .insert(apikey_entry("sk-caller", &["my-video-model"]));
+        let state = build_state(snap, hub);
+
+        let body = serde_json::json!({
+            "model": "my-video-model",
+            "input": {"prompt": "a cardboard city at night"},
+            "parameters": {"resolution": "720P"}
+        });
+        let make_req = || {
+            Request::builder()
+                .method("POST")
+                .uri("/passthrough/openai/api/v1/services/aigc/video-generation/video-synthesis")
+                .header("authorization", "Bearer sk-caller")
+                .header("content-type", "application/json")
+                .body(Body::from(body.to_string()))
+                .unwrap()
+        };
+
+        // First request consumes the model's only RPM slot.
+        let resp = run(build_router(state.clone()), make_req()).await;
+        assert_eq!(resp.status(), StatusCode::OK);
+
+        // Pre-fix: 200 again (model layer skipped on passthrough).
+        // Post-fix: 429 — the body's `model` resolved the configured
+        // Model and its rpm=1 cap now gates the tunnel.
+        let resp = run(build_router(state.clone()), make_req()).await;
+        assert_eq!(
+            resp.status(),
+            StatusCode::TOO_MANY_REQUESTS,
+            "passthrough must enforce the body model's rate limit (api7/AISIX-Cloud#1116)",
+        );
+        let body_bytes = to_bytes(resp.into_body(), 1024).await.unwrap();
+        let v: serde_json::Value = serde_json::from_slice(&body_bytes).unwrap();
+        assert_eq!(v["error"]["type"], "rate_limit_exceeded");
+    }
+
+    #[tokio::test]
+    async fn passthrough_unregistered_or_absent_body_model_keeps_key_layer_only() {
+        let upstream = MockServer::start().await;
+        Mock::given(method("POST"))
+            .and(path("/anything"))
+            .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({"ok": true})))
+            .mount(&upstream)
+            .await;
+
+        let hub = Arc::new(Hub::new());
+        hub.register_specialized("openai", Arc::new(openai_test_bridge()));
+        // Model has rpm=1, but requests below never name it — its bucket
+        // must stay untouched. The API key carries rpm=2 to pin that the
+        // key layer still gates the tunnel (and 429s on the 3rd call).
+        let snap = new_snap(&upstream.uri());
+        snap.models.insert(model_entry_with_rate_limit(
+            "my-video-model",
+            serde_json::json!({"rpm": 1}),
+        ));
+        snap.apikeys.insert(apikey_entry_with_limits(
+            "sk-caller",
+            &["my-video-model"],
+            Some(serde_json::json!({"rpm": 2})),
+        ));
+        let state = build_state(snap, hub);
+
+        let make_req = |body: &'static str| {
+            Request::builder()
+                .method("POST")
+                .uri("/passthrough/openai/anything")
+                .header("authorization", "Bearer sk-caller")
+                .header("content-type", "application/json")
+                .body(Body::from(body))
+                .unwrap()
+        };
+
+        // Unregistered model name → no model-layer reservation, passes.
+        let resp = run(
+            build_router(state.clone()),
+            make_req(r#"{"model":"not-a-configured-model","input":"x"}"#),
+        )
+        .await;
+        assert_eq!(resp.status(), StatusCode::OK);
+
+        // Non-JSON body → tolerated, no model-layer reservation.
+        let resp = run(build_router(state.clone()), make_req("plain text body")).await;
+        assert_eq!(resp.status(), StatusCode::OK);
+
+        // Third call trips the KEY-level rpm=2 — the request-level
+        // layers keep gating the tunnel exactly as before the fix.
+        let resp = run(
+            build_router(state.clone()),
+            make_req(r#"{"model":"not-a-configured-model","input":"x"}"#),
+        )
+        .await;
+        assert_eq!(
+            resp.status(),
+            StatusCode::TOO_MANY_REQUESTS,
+            "key-level rate limit must keep applying to passthrough",
+        );
+    }
+
+    /// A `model`-scope RateLimitPolicy row (no inline rate_limit on the
+    /// Model) must gate the tunnel too — the policy path matches by the
+    /// resolved entry id, which is only exercised when the body model
+    /// lookup propagates it.
+    #[tokio::test]
+    async fn passthrough_enforces_model_scope_policy_from_body_model_field() {
+        let upstream = MockServer::start().await;
+        Mock::given(method("POST"))
+            .and(path("/anything"))
+            .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({"ok": true})))
+            .mount(&upstream)
+            .await;
+
+        let hub = Arc::new(Hub::new());
+        hub.register_specialized("openai", Arc::new(openai_test_bridge()));
+        let snap = new_snap(&upstream.uri());
+        snap.models.insert(model_entry("my-video-model"));
+        snap.rate_limit_policies.insert(ResourceEntry::new(
+            "pol-1",
+            serde_json::from_value(serde_json::json!({
+                "name": "video-cap",
+                "scope": "model",
+                "scope_ref": "model-id-1",
+                "window": "minute",
+                "max_requests": 1
+            }))
+            .unwrap(),
+            1,
+        ));
+        snap.apikeys
+            .insert(apikey_entry("sk-caller", &["my-video-model"]));
+        let state = build_state(snap, hub);
+
+        let make_req = || {
+            Request::builder()
+                .method("POST")
+                .uri("/passthrough/openai/anything")
+                .header("authorization", "Bearer sk-caller")
+                .header("content-type", "application/json")
+                .body(Body::from(r#"{"model":"my-video-model","input":"x"}"#))
+                .unwrap()
+        };
+
+        let resp = run(build_router(state.clone()), make_req()).await;
+        assert_eq!(resp.status(), StatusCode::OK);
+
+        let resp = run(build_router(state.clone()), make_req()).await;
+        assert_eq!(
+            resp.status(),
+            StatusCode::TOO_MANY_REQUESTS,
+            "model-scope policy must gate passthrough via the body model",
+        );
+    }
+
+    /// The tunnel forwards bodies verbatim, so callers typically name the
+    /// provider-native id (`model_name`), not the gateway alias
+    /// (`display_name`). The limit must bind either way, and the bucket is
+    /// keyed by the alias so tunnel and typed traffic share one budget.
+    #[tokio::test]
+    async fn passthrough_matches_provider_native_model_name_for_rate_limit() {
+        let upstream = MockServer::start().await;
+        Mock::given(method("POST"))
+            .and(path("/anything"))
+            .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({"ok": true})))
+            .mount(&upstream)
+            .await;
+
+        let hub = Arc::new(Hub::new());
+        hub.register_specialized("openai", Arc::new(openai_test_bridge()));
+        let snap = new_snap(&upstream.uri());
+        snap.models.insert(model_entry_named(
+            "ali-video-alias",
+            "happyhorse-1.1-t2v",
+            serde_json::json!({"rpm": 1}),
+        ));
+        snap.apikeys
+            .insert(apikey_entry("sk-caller", &["ali-video-alias"]));
+        let state = build_state(snap, hub);
+
+        // Caller names the upstream id, not the alias — the alias's cap
+        // must still bind.
+        let make_req = || {
+            Request::builder()
+                .method("POST")
+                .uri("/passthrough/openai/anything")
+                .header("authorization", "Bearer sk-caller")
+                .header("content-type", "application/json")
+                .body(Body::from(r#"{"model":"happyhorse-1.1-t2v","input":"x"}"#))
+                .unwrap()
+        };
+
+        let resp = run(build_router(state.clone()), make_req()).await;
+        assert_eq!(resp.status(), StatusCode::OK);
+
+        let resp = run(build_router(state.clone()), make_req()).await;
+        assert_eq!(
+            resp.status(),
+            StatusCode::TOO_MANY_REQUESTS,
+            "provider-native model_name must resolve the Model's rate limit",
+        );
+    }
+
+    /// A same-named Model registered under a DIFFERENT provider must not
+    /// be charged for this tunnel's traffic: the body name only matches
+    /// within the addressed provider.
+    #[tokio::test]
+    async fn passthrough_same_named_model_of_other_provider_is_not_limited() {
+        let upstream = MockServer::start().await;
+        Mock::given(method("POST"))
+            .and(path("/anything"))
+            .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({"ok": true})))
+            .mount(&upstream)
+            .await;
+
+        let hub = Arc::new(Hub::new());
+        hub.register_specialized("openai", Arc::new(openai_test_bridge()));
+        let snap = new_snap(&upstream.uri());
+        // Credential-lending model for the addressed provider — unlimited.
+        snap.models.insert(model_entry("openai-lender"));
+        // Same body-name model under another provider, rpm=1. Its bucket
+        // must stay untouched by /passthrough/openai traffic.
+        let cross: Model = serde_json::from_str(&format!(
+            r#"{{
+                "display_name": "cross-model",
+                "provider": "anthropic",
+                "model_name": "cross-model",
+                "provider_key_id": "{PK_ID}",
+                "rate_limit": {{"rpm": 1}}
+            }}"#
+        ))
+        .unwrap();
+        snap.models
+            .insert(ResourceEntry::new("model-id-cross", cross, 1));
+        snap.apikeys
+            .insert(apikey_entry("sk-caller", &["openai-lender", "cross-model"]));
+        let state = build_state(snap, hub);
+
+        let make_req = || {
+            Request::builder()
+                .method("POST")
+                .uri("/passthrough/openai/anything")
+                .header("authorization", "Bearer sk-caller")
+                .header("content-type", "application/json")
+                .body(Body::from(r#"{"model":"cross-model","input":"x"}"#))
+                .unwrap()
+        };
+
+        // Both calls pass: the anthropic model's rpm=1 bucket is never
+        // drawn from by the openai tunnel.
+        for _ in 0..2 {
+            let resp = run(build_router(state.clone()), make_req()).await;
+            assert_eq!(
+                resp.status(),
+                StatusCode::OK,
+                "cross-provider same-named model must not gate this tunnel",
+            );
+        }
+    }
+
     /// Regression for issue #108: streaming chat used to commit
     /// `0` tokens up front and never look at the upstream's terminal
     /// usage frame. TPM caps were silently bypassed for all streaming