1use std::fmt;
2use std::net::IpAddr;
3use std::sync::Arc;
4use std::time::Duration;
5
6use http::Method;
7use serde_json::Value;
8use url::{Host, Url};
9
10use crate::adapter::{
11 push_string, push_text_blocks, walk_all_strings, AdapterContract, PiiSurface, ProviderAdapter,
12 SessionPolicy, SessionRegistryConfig, SseEvent,
13};
14use crate::codec::CodecLimits;
15use crate::codecs::anthropic::AnthropicMessagesCodec;
16use crate::error::{DirectProxyError, ProxyErrorCode, ProxyErrorPhase};
17use crate::principal::PrincipalResolver;
18
19pub const DEFAULT_ANTHROPIC_UPSTREAM: &str = "https://api.anthropic.com";
20pub const DEFAULT_ANTHROPIC_VERSION: &str = "2023-06-01";
21
22const DEFAULT_CONNECT_TIMEOUT: Duration = Duration::from_secs(5);
23const DEFAULT_REQUEST_TIMEOUT: Duration = Duration::from_secs(120);
24const DEFAULT_TOTAL_TIMEOUT: Duration = Duration::from_secs(180);
25const DEFAULT_PING_INTERVAL: Duration = Duration::from_secs(10);
26const MAX_HEADER_COUNT: usize = 64;
27const MAX_HEADER_NAME_BYTES: usize = 128;
28const MAX_HEADER_VALUE_BYTES: usize = 8 * 1024;
29const MAX_HEADER_BYTES: usize = 64 * 1024;
30
31#[derive(Clone)]
32#[non_exhaustive]
33pub struct AnthropicAdapter {
34 upstream: Url,
35 codec: AnthropicMessagesCodec,
36 session_policy: SessionPolicy,
37 allowed_versions: Arc<[String]>,
38 allowed_betas: Arc<[String]>,
39 codec_limits: CodecLimits,
40 connect_timeout: Duration,
41 request_timeout: Duration,
42 total_timeout: Duration,
43 ping_interval: Duration,
44 max_headers: usize,
45 max_header_name_bytes: usize,
46 max_header_value_bytes: usize,
47 max_header_bytes: usize,
48 principal_resolver: Option<Arc<dyn PrincipalResolver>>,
49}
50
51impl fmt::Debug for AnthropicAdapter {
52 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
53 formatter
54 .debug_struct("AnthropicAdapter")
55 .field("session_policy", &self.session_policy)
56 .field("allowed_version_count", &self.allowed_versions.len())
57 .field("allowed_beta_count", &self.allowed_betas.len())
58 .field("codec_limits", &self.codec_limits)
59 .field("has_principal_resolver", &self.principal_resolver.is_some())
60 .finish()
61 }
62}
63
64impl AnthropicAdapter {
65 pub fn new(upstream: Url) -> Self {
66 Self {
67 upstream,
68 codec: AnthropicMessagesCodec,
69 session_policy: SessionPolicy::EphemeralSingleRequest,
70 allowed_versions: Arc::from([DEFAULT_ANTHROPIC_VERSION.to_string()]),
71 allowed_betas: Arc::<[String]>::from([]),
72 codec_limits: CodecLimits::default(),
73 connect_timeout: DEFAULT_CONNECT_TIMEOUT,
74 request_timeout: DEFAULT_REQUEST_TIMEOUT,
75 total_timeout: DEFAULT_TOTAL_TIMEOUT,
76 ping_interval: DEFAULT_PING_INTERVAL,
77 max_headers: MAX_HEADER_COUNT,
78 max_header_name_bytes: MAX_HEADER_NAME_BYTES,
79 max_header_value_bytes: MAX_HEADER_VALUE_BYTES,
80 max_header_bytes: MAX_HEADER_BYTES,
81 principal_resolver: None,
82 }
83 }
84
85 #[must_use]
86 pub fn builder(upstream: Url) -> AnthropicAdapterBuilder {
87 AnthropicAdapterBuilder::new(upstream)
88 }
89
90 #[must_use]
91 pub fn allowed_versions(&self) -> &[String] {
92 &self.allowed_versions
93 }
94
95 #[must_use]
96 pub fn allowed_betas(&self) -> &[String] {
97 &self.allowed_betas
98 }
99
100 pub(crate) fn validate_upstream_origin(&self) -> Result<(), DirectProxyError> {
101 validate_upstream_origin(&self.upstream)
102 }
103
104 pub(crate) const fn connect_timeout(&self) -> Duration {
105 self.connect_timeout
106 }
107
108 pub(crate) const fn request_timeout(&self) -> Duration {
109 self.request_timeout
110 }
111
112 pub(crate) const fn total_timeout(&self) -> Duration {
113 self.total_timeout
114 }
115
116 pub(crate) const fn ping_interval(&self) -> Duration {
117 self.ping_interval
118 }
119
120 pub(crate) const fn header_limits(&self) -> (usize, usize, usize, usize) {
121 (
122 self.max_headers,
123 self.max_header_name_bytes,
124 self.max_header_value_bytes,
125 self.max_header_bytes,
126 )
127 }
128
129 pub(crate) fn principal_resolver(&self) -> Option<&Arc<dyn PrincipalResolver>> {
130 self.principal_resolver.as_ref()
131 }
132}
133
134pub struct AnthropicAdapterBuilder {
135 adapter: AnthropicAdapter,
136}
137
138impl fmt::Debug for AnthropicAdapterBuilder {
139 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
140 formatter
141 .debug_struct("AnthropicAdapterBuilder")
142 .field("adapter", &self.adapter)
143 .finish()
144 }
145}
146
147impl AnthropicAdapterBuilder {
148 fn new(upstream: Url) -> Self {
149 let mut adapter = AnthropicAdapter::new(upstream);
150 adapter.session_policy =
151 SessionPolicy::OpaqueHeaderContinuity(SessionRegistryConfig::default());
152 Self { adapter }
153 }
154
155 #[must_use]
156 pub fn session_registry_config(mut self, config: SessionRegistryConfig) -> Self {
157 self.adapter.session_policy = SessionPolicy::OpaqueHeaderContinuity(config);
158 self
159 }
160
161 #[must_use]
162 pub fn codec_limits(mut self, limits: CodecLimits) -> Self {
163 self.adapter.codec_limits = limits;
164 self
165 }
166
167 pub fn connect_timeout(mut self, timeout: Duration) -> Result<Self, DirectProxyError> {
168 validate_timeout(timeout, DEFAULT_CONNECT_TIMEOUT)?;
169 self.adapter.connect_timeout = timeout;
170 Ok(self)
171 }
172
173 pub fn request_timeout(mut self, timeout: Duration) -> Result<Self, DirectProxyError> {
174 validate_timeout(timeout, DEFAULT_REQUEST_TIMEOUT)?;
175 self.adapter.request_timeout = timeout;
176 Ok(self)
177 }
178
179 pub fn total_timeout(mut self, timeout: Duration) -> Result<Self, DirectProxyError> {
180 validate_timeout(timeout, DEFAULT_TOTAL_TIMEOUT)?;
181 self.adapter.total_timeout = timeout;
182 Ok(self)
183 }
184
185 pub fn ping_interval(mut self, interval: Duration) -> Result<Self, DirectProxyError> {
186 if !(Duration::from_secs(1)..=Duration::from_secs(30)).contains(&interval) {
187 return Err(
188 ProxyErrorCode::ProxyConfiguration.error(ProxyErrorPhase::UpstreamConfiguration)
189 );
190 }
191 self.adapter.ping_interval = interval;
192 Ok(self)
193 }
194
195 pub fn max_headers(mut self, value: usize) -> Result<Self, DirectProxyError> {
196 validate_lowered_limit(value, MAX_HEADER_COUNT)?;
197 self.adapter.max_headers = value;
198 Ok(self)
199 }
200
201 pub fn max_header_name_bytes(mut self, value: usize) -> Result<Self, DirectProxyError> {
202 validate_lowered_limit(value, MAX_HEADER_NAME_BYTES)?;
203 self.adapter.max_header_name_bytes = value;
204 Ok(self)
205 }
206
207 pub fn max_header_value_bytes(mut self, value: usize) -> Result<Self, DirectProxyError> {
208 validate_lowered_limit(value, MAX_HEADER_VALUE_BYTES)?;
209 self.adapter.max_header_value_bytes = value;
210 Ok(self)
211 }
212
213 pub fn max_header_bytes(mut self, value: usize) -> Result<Self, DirectProxyError> {
214 validate_lowered_limit(value, MAX_HEADER_BYTES)?;
215 self.adapter.max_header_bytes = value;
216 Ok(self)
217 }
218
219 pub fn allow_version(mut self, version: impl Into<String>) -> Result<Self, DirectProxyError> {
220 let version = version.into();
221 validate_allowlist_value(&version)?;
222 let mut values = self.adapter.allowed_versions.to_vec();
223 if !values.contains(&version) {
224 values.push(version);
225 }
226 self.adapter.allowed_versions = values.into();
227 Ok(self)
228 }
229
230 pub fn allow_beta(mut self, beta: impl Into<String>) -> Result<Self, DirectProxyError> {
231 let beta = beta.into();
232 validate_allowlist_value(&beta)?;
233 let mut values = self.adapter.allowed_betas.to_vec();
234 if !values.contains(&beta) {
235 values.push(beta);
236 }
237 self.adapter.allowed_betas = values.into();
238 Ok(self)
239 }
240
241 #[must_use]
242 pub fn principal_resolver(mut self, resolver: Arc<dyn PrincipalResolver>) -> Self {
243 self.adapter.principal_resolver = Some(resolver);
244 self
245 }
246
247 pub fn build(self) -> Result<AnthropicAdapter, DirectProxyError> {
248 self.adapter.validate_upstream_origin()?;
249 validate_codec_limits(self.adapter.codec_limits)?;
250 if self.adapter.connect_timeout > self.adapter.request_timeout
251 || self.adapter.request_timeout > self.adapter.total_timeout
252 {
253 return Err(configuration_error());
254 }
255 Ok(self.adapter)
256 }
257}
258
259fn validate_timeout(timeout: Duration, maximum: Duration) -> Result<(), DirectProxyError> {
260 if timeout.is_zero() || timeout > maximum {
261 Err(configuration_error())
262 } else {
263 Ok(())
264 }
265}
266
267fn validate_lowered_limit(value: usize, maximum: usize) -> Result<(), DirectProxyError> {
268 if value == 0 || value > maximum {
269 Err(configuration_error())
270 } else {
271 Ok(())
272 }
273}
274
275fn validate_codec_limits(limits: CodecLimits) -> Result<(), DirectProxyError> {
276 let ceilings = CodecLimits::default();
277 for (value, maximum) in [
278 (limits.max_body_bytes(), ceilings.max_body_bytes()),
279 (limits.max_json_depth(), ceilings.max_json_depth()),
280 (limits.max_json_nodes(), ceilings.max_json_nodes()),
281 (limits.max_string_bytes(), ceilings.max_string_bytes()),
282 (limits.max_number_bytes(), ceilings.max_number_bytes()),
283 (limits.max_sse_line_bytes(), ceilings.max_sse_line_bytes()),
284 (limits.max_sse_frame_bytes(), ceilings.max_sse_frame_bytes()),
285 (limits.max_sse_events(), ceilings.max_sse_events()),
286 (limits.max_sse_indices(), ceilings.max_sse_indices()),
287 (
288 limits.max_sse_accumulator_bytes(),
289 ceilings.max_sse_accumulator_bytes(),
290 ),
291 ] {
292 validate_lowered_limit(value, maximum)?;
293 }
294
295 let body = limits.max_body_bytes();
296 if limits.max_sse_line_bytes() > limits.max_sse_frame_bytes()
297 || limits.max_sse_line_bytes() > body
298 || limits.max_sse_frame_bytes() > body
299 || limits.max_string_bytes() > body
300 || limits.max_number_bytes() > body
301 || limits.max_sse_accumulator_bytes() > body
302 {
303 return Err(configuration_error());
304 }
305
306 Ok(())
307}
308
309fn validate_allowlist_value(value: &str) -> Result<(), DirectProxyError> {
310 if value.is_empty()
311 || value.len() > MAX_HEADER_VALUE_BYTES
312 || value
313 .bytes()
314 .any(|byte| !matches!(byte, 0x21..=0x7e) || byte == b',')
315 {
316 Err(ProxyErrorCode::HeaderRejected.error(ProxyErrorPhase::UpstreamConfiguration))
317 } else {
318 Ok(())
319 }
320}
321
322fn validate_upstream_origin(upstream: &Url) -> Result<(), DirectProxyError> {
323 let safe_authority = upstream.username().is_empty()
324 && upstream.password().is_none()
325 && upstream.query().is_none()
326 && upstream.fragment().is_none()
327 && upstream.path() == "/"
328 && upstream.host().is_some();
329 let safe_scheme = match (upstream.scheme(), upstream.host()) {
330 ("https", Some(_)) => true,
331 ("http", Some(Host::Ipv4(address))) => IpAddr::V4(address).is_loopback(),
332 ("http", Some(Host::Ipv6(address))) => IpAddr::V6(address).is_loopback(),
333 _ => false,
334 };
335 if safe_authority && safe_scheme {
336 Ok(())
337 } else {
338 Err(ProxyErrorCode::InvalidUpstreamUrl.error(ProxyErrorPhase::UpstreamConfiguration))
339 }
340}
341
342fn configuration_error() -> DirectProxyError {
343 ProxyErrorCode::ProxyConfiguration.error(ProxyErrorPhase::UpstreamConfiguration)
344}
345
346#[async_trait::async_trait]
347impl ProviderAdapter for AnthropicAdapter {
348 fn contract(&self) -> AdapterContract<'_> {
349 AdapterContract::codec(&self.codec, self.session_policy, self.codec_limits)
350 }
351
352 fn name(&self) -> &'static str {
353 "anthropic"
354 }
355
356 fn matches_path(&self, method: &Method, path: &str) -> bool {
357 method == Method::POST && path == "/v1/messages"
358 }
359
360 fn upstream_base(&self) -> &Url {
361 &self.upstream
362 }
363
364 fn request_pii_surfaces<'a>(&self, body: &'a mut Value) -> Vec<PiiSurface<'a>> {
365 let mut surfaces = Vec::new();
366 if let Value::Object(root) = body {
367 for (key, value) in root {
368 match key.as_str() {
369 "system" => push_text_blocks(&mut surfaces, "system", value),
370 "messages" => {
371 if let Value::Array(messages) = value {
372 for (index, message) in messages.iter_mut().enumerate() {
373 if let Value::Object(message) = message {
374 for (message_key, message_value) in message {
375 if message_key == "content" {
376 collect_content_blocks(
377 &mut surfaces,
378 &format!("messages[{index}].content"),
379 message_value,
380 );
381 }
382 }
383 }
384 }
385 }
386 }
387 _ => {}
388 }
389 }
390 }
391 surfaces
392 }
393
394 fn response_pii_surfaces<'a>(&self, body: &'a mut Value) -> Vec<PiiSurface<'a>> {
395 let mut surfaces = Vec::new();
396 if let Value::Object(root) = body {
397 if let Some(content) = root.get_mut("content") {
398 collect_content_blocks(&mut surfaces, "content", content);
399 }
400 }
401 surfaces
402 }
403
404 fn sse_event_pii_surfaces<'a>(&self, event: &'a mut SseEvent) -> Vec<PiiSurface<'a>> {
405 let mut surfaces = Vec::new();
406 if let Value::Object(root) = &mut event.data {
407 if matches!(root.get("type"), Some(Value::String(kind)) if kind == "content_block_delta")
408 {
409 if let Some(Value::Object(delta)) = root.get_mut("delta") {
410 for (key, value) in delta {
411 if let Value::String(text) = value {
412 match key.as_str() {
413 "text" => surfaces.push(PiiSurface {
414 field_path: "delta.text".to_string(),
415 text,
416 }),
417 "partial_json" => surfaces.push(PiiSurface {
418 field_path: "delta.partial_json".to_string(),
419 text,
420 }),
421 _ => {}
422 }
423 }
424 }
425 }
426 }
427 }
428 surfaces
429 }
430}
431
432fn collect_content_blocks<'a>(
433 surfaces: &mut Vec<PiiSurface<'a>>,
434 prefix: &str,
435 content: &'a mut Value,
436) {
437 match content {
438 Value::String(_) => push_string(surfaces, prefix, content),
439 Value::Array(blocks) => {
440 for (index, block) in blocks.iter_mut().enumerate() {
441 let Value::Object(block) = block else {
442 continue;
443 };
444 match block.get("type").and_then(Value::as_str) {
445 Some("text") => {
446 if let Some(Value::String(text)) = block.get_mut("text") {
447 surfaces.push(PiiSurface {
448 field_path: format!("{prefix}[{index}].text"),
449 text,
450 });
451 }
452 }
453 Some("tool_use") => {
454 if let Some(input) = block.get_mut("input") {
455 walk_all_strings(surfaces, format!("{prefix}[{index}].input"), input);
456 }
457 }
458 Some("tool_result") => {
459 if let Some(result) = block.get_mut("content") {
460 collect_content_blocks(
461 surfaces,
462 &format!("{prefix}[{index}].content"),
463 result,
464 );
465 }
466 }
467 Some("image") | None | Some(_) => {}
468 }
469 }
470 }
471 _ => {}
472 }
473}