feat(sources): add custom auth strategy for components with HTTP server

[esensar]

Summary

This adds custom auth strategy for components with HTTP server (http_server, datadog_agent, opentelemetry, prometheus) besides the default basic auth. This is a breaking change because strategy is now required for auth - for existing configurations strategy: "basic" needs to be added.

Change Type

• Bug fix
• New feature
• Non-functional (chore, refactoring, docs)
• Performance

Is this a breaking change?

• Yes
• No

How did you test this PR?

Besides the tests added to the codebase, I ran basic tests with http_server source component:

sources:
  http_server_source:
    type: "http_server"
    address: "0.0.0.0:80"
    auth:
      strategy: "custom"
      source: |-
        .headers.authorization == "test"

sinks:
  console:
    inputs: ["http_server_source"]
    target: "stdout"
    type: "console"
    acknowledgements:
      enabled: false
    encoding:
      codec: "json"

Tested by making calls via curl:

$ curl -X POST vector:80
{"code":401,"message":"Auth failed"}
$ curl -X POST vector:80 -H "Authorization: test"

Does this PR include user facing changes?

• Yes. Please add a changelog fragment based on our guidelines.
• No. A maintainer will apply the "no-changelog" label to this PR.

Checklist

• Please read our Vector contributor resources.
  • make check-all is a good command to run locally. This check is
    defined here. Some of these
    checks might not be relevant to your PR. For Rust changes, at the very least you should run:
    • cargo fmt --all
    • cargo clippy --workspace --all-targets -- -D warnings
    • cargo nextest run --workspace (alternatively, you can run cargo test --all)
• If this PR introduces changes Vector dependencies (modifies Cargo.lock), please
  run dd-rust-license-tool write to regenerate the license inventory and commit the changes (if any). More details here.

References

Related: #22213

[esensar]

I have made this a breaking change, requiring explicit strategy, for consistency (I have had a similar situation in a previous contribution: #19892 (comment)). Let me know if you want me to add untagged for serde deserialization, to make this a non-breaking change.

[pront]

Seems like we have another use case for serde-rs/serde#2231.

Related: #22212 (comment)

💭 Thinking how to avoid breaking behavior for users. I will play locally with a custom deserializer and come back to you.

[pront]

Change to HttpServerAuthConfig to:

#[configurable_component(no_deser)]

Custom deserializer (mostly AI generated):

// Custom deserializer to default `strategy` to `basic`
impl<'de> Deserialize<'de> for HttpServerAuthConfig {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: Deserializer<'de>,
    {
        struct HttpServerAuthConfigVisitor;

        impl<'de> Visitor<'de> for HttpServerAuthConfigVisitor {
            type Value = HttpServerAuthConfig;

            fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
                formatter.write_str("a valid authentication strategy (basic or custom)")
            }

            fn visit_map<A>(self, mut map: A) -> Result<HttpServerAuthConfig, A::Error>
            where
                A: MapAccess<'de>,
            {
                let mut strategy: Option<String> = None;
                let mut username: Option<String> = None;
                let mut password: Option<String> = None;
                let mut source: Option<String> = None;

                while let Some(key) = map.next_key::<String>()? {
                    match key.as_str() {
                        "strategy" => {
                            if strategy.is_some() {
                                return Err(de::Error::duplicate_field("strategy"));
                            }
                            strategy = Some(map.next_value()?);
                        }
                        "username" => {
                            if username.is_some() {
                                return Err(de::Error::duplicate_field("username"));
                            }
                            username = Some(map.next_value()?);
                        }
                        "password" => {
                            if password.is_some() {
                                return Err(de::Error::duplicate_field("password"));
                            }
                            password = Some(map.next_value()?);
                        }
                        "source" => {
                            if source.is_some() {
                                return Err(de::Error::duplicate_field("source"));
                            }
                            source = Some(map.next_value()?);
                        }
                        _ => {
                            return Err(de::Error::unknown_field(
                                &key,
                                &["strategy", "username", "password", "source"],
                            ));
                        }
                    }
                }

                // Default to "basic" if strategy is missing
                let strategy = strategy.unwrap_or_else(|| "basic".to_string());

                match strategy.as_str() {
                    "basic" => {
                        let username = username.ok_or_else(|| de::Error::missing_field("username"))?;
                        let password = password.ok_or_else(|| de::Error::missing_field("password"))?;
                        Ok(HttpServerAuthConfig::Basic {
                            username,
                            password: SensitiveString::from(password),
                        })
                    }
                    "custom" => {
                        let source = source.ok_or_else(|| de::Error::missing_field("source"))?;
                        Ok(HttpServerAuthConfig::Custom { source })
                    }
                    _ => Err(de::Error::unknown_variant(&strategy, &["basic", "custom"])),
                }
            }
        }

        deserializer.deserialize_map(HttpServerAuthConfigVisitor)
    }
}

Tried with, config 1:

sources:
  s0:
    type: http_server
    address: 0.0.0.0:80
    auth:
      # not specifying strategy
      username: foo
      password: bar
...

..and config 2:

sources:
  s0:
    type: http_server
    address: 0.0.0.0:80
    auth:
      strategy: custom
      source: "true"

[pront]

Thanks @esensar! Happy that we can avoid breaking existing behavior. This PR is almost there.

[pront]

Is this no longer needed?

[esensar]

Authorization is now handled by handle_auth which takes in all headers, and they are already picked up below (warp::header::headers_cloned()).