Repository: linkerd/linkerd2
Pull request #6725 (2021-08-23 17:15:33)

Bump serde_yaml from 0.8.18 to 0.8.19 in /policy-controller
Bumps [serde_yaml](https://github.com/dtolnay/serde-yaml) from 0.8.18 to 0.8.19.
<details>
<summary>Release notes</summary>
<p><em>Sourced from <a href="https://github.com/dtolnay/serde-yaml/releases">serde_yaml's releases</a>.</em></p>
<blockquote>
<h2>0.8.19</h2>
<ul>
<li>Add an <code>Entry</code> API for serde_yaml::Mapping (<a href="https://docs.rs/serde_yaml/0.8.19/serde_yaml/struct.Mapping.html#method.entry">https://docs.rs/serde_yaml/0.8.19/serde_yaml/struct.Mapping.html#method.entry</a>)</li>
</ul>
</blockquote>
</details>
<details>
<summary>Commits</summary>
<ul>
<li><a href="https://github.com/dtolnay/serde-yaml/commit/e5c11139e5c47d55ffb71cbd20aee77c0cc75aaa"><code>e5c1113</code></a> Release 0.8.19</li>
<li><a href="https://github.com/dtolnay/serde-yaml/commit/93a7ed1420c3a6eac147845245a2298038121538"><code>93a7ed1</code></a> Merge pull request <a href="https://github-redirect.dependabot.com/dtolnay/serde-yaml/issues/209">#209</a> from dtolnay/entry</li>
<li><a href="https://github.com/dtolnay/serde-yaml/commit/0b9a0af92db9cb2991105fb389717e6b89829536"><code>0b9a0af</code></a> Silence doc_markdown Clippy pedantic lint</li>
<li><a href="https://github.com/dtolnay/serde-yaml/commit/d6c40e68c4c043480258d42cde0c917e37184cdc"><code>d6c40e6</code></a> Add an Entry API for Mapping</li>
<li><a href="https://github.com/dtolnay/serde-yaml/commit/81505bf4579fc71fe3f49887fedda8e552331900"><code>81505bf</code></a> Merge pull request <a href="https://github-redirect.dependabot.com/dtolnay/serde-yaml/issues/208">#208</a> from dtolnay/indexmap</li>
<li><a href="https://github.com/dtolnay/serde-yaml/commit/284af2f35248178a7cd5f6b7a9421609f4309a04"><code>284af2f</code></a> Ignore derive_hash_xor_eq Clippy lint</li>
<li><a href="https://github.com/dtolnay/serde-yaml/commit/7148f9b0525193a07034d21f967f902b2ea16f67"><code>7148f9b</code></a> Ignore items_after_statements Clippy pedantic lint</li>
<li><a href="https://github.com/dtolnay/serde-yaml/commit/aed90c2cf8d9c15027c7f5a67e7a42893abbc883"><code>aed90c2</code></a> Ignore if_not_else Clippy pedantic lint</li>
<li><a href="https://github.com/dtolnay/serde-yaml/commit/6e432a2681462ea0bd9f343a84de534c6d5885ae"><code>6e432a2</code></a> Ignore redundant_else Clippy pedantic lint</li>
<li><a href="https://github.com/dtolnay/serde-yaml/commit/6edeb29eba8ff6f6c6d5b607ce953895bbf62c19"><code>6edeb29</code></a> Raise minimum compiler to 1.38</li>
<li>Additional commits viewable in <a href="https://github.com/dtolnay/serde-yaml/compare/0.8.18...0.8.19">compare view</a></li>
</ul>
</details>
<br />


[![Dependabot compatibility score](https://dependabot-badges.githubapp.com/badges/compatibility_score?dependency-name=serde_yaml&package-manager=cargo&previous-version=0.8.18&new-version=0.8.19)](https://docs.github.com/en/github/managing-security-vulnerabilities/about-dependabot-security-updates#about-compatibility-scores)

Dependabot will resolve any conflicts with this PR as long as you don't alter it yourself. You can also trigger a rebase manually by commenting `@dependabot rebase`.

[//]: # (dependabot-automerge-start)
[//]: # (dependabot-automerge-end)

---

<details>
<summary>Dependabot commands and options</summary>
<br />

You can trigger Dependabot actions by commenting on this PR:
- `@dependabot rebase` will rebase this PR
- `@dependabot recreate` will recreate this PR, overwriting any edits that have been made to it
- `@dependabot merge` will merge this PR after your CI passes on it
- `@dependabot squash and merge` will squash and merge this PR after your CI passes on it
- `@dependabot cancel merge` will cancel a previously requested merge and block automerging
- `@dependabot reopen` will reopen this PR if it is closed
- `@dependabot close` will close this PR and stop Dependabot recreating it. You can achieve the same result by closing it manually
- `@dependabot ignore this major version` will close this PR and stop Dependabot creating any more for this major version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this minor version` will close this PR and stop Dependabot creating any more for this minor version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this dependency` will close this PR and stop Dependabot creating any more for this dependency (unless you reopen the PR or upgrade to it yourself)


</details>


Patch:
diff --git a/charts/linkerd2/templates/destination.yaml b/charts/linkerd2/templates/destination.yaml
index 6264988cc..78f2096d0 100644
--- a/charts/linkerd2/templates/destination.yaml
+++ b/charts/linkerd2/templates/destination.yaml
@@ -226,7 +226,7 @@ spec:
         - --grpc-addr=0.0.0.0:8090
         - --cluster-networks={{.Values.clusterNetworks}}
         - --identity-domain={{.Values.identityTrustDomain | default .Values.clusterDomain}}
-        - --default-allow={{.Values.policyController.defaultAllowPolicy}}
+        - --default-policy={{.Values.policyController.defaultAllowPolicy}}
         env:
         - name: RUST_LOG
           value: {{.Values.policyController.logLevel | default "linkerd=info,warn" | quote}}
diff --git a/policy-controller/Cargo.lock b/policy-controller/Cargo.lock
index ba5381152..b999a860a 100644
--- a/policy-controller/Cargo.lock
+++ b/policy-controller/Cargo.lock
@@ -1454,9 +1454,9 @@ dependencies = [
 
 [[package]]
 name = "serde"
-version = "1.0.127"
+version = "1.0.128"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f03b9878abf6d14e6779d3f24f07b2cfa90352cfec4acc5aab8f1ac7f146fae8"
+checksum = "1056a0db1978e9dbf0f6e4fca677f6f9143dc1c19de346f22cac23e422196834"
 dependencies = [
  "serde_derive",
 ]
@@ -1473,9 +1473,9 @@ dependencies = [
 
 [[package]]
 name = "serde_derive"
-version = "1.0.127"
+version = "1.0.128"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a024926d3432516606328597e0f224a51355a493b49fdd67e9209187cbe55ecc"
+checksum = "13af2fbb8b60a8950d6c72a56d2095c28870367cc8e10c55e9745bac4995a2c4"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -1507,12 +1507,12 @@ dependencies = [
 
 [[package]]
 name = "serde_yaml"
-version = "0.8.18"
+version = "0.8.19"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "039ba818c784248423789eec090aab9fb566c7b94d6ebbfa1814a9fd52c8afb2"
+checksum = "6375dbd828ed6964c3748e4ef6d18e7a175d408ffe184bca01698d0c73f915a9"
 dependencies = [
  "dtoa",
- "linked-hash-map",
+ "indexmap",
  "serde",
  "yaml-rust",
 ]
diff --git a/policy-controller/core/src/lib.rs b/policy-controller/core/src/lib.rs
index ebbdd7a4e..1e26a47c5 100644
--- a/policy-controller/core/src/lib.rs
+++ b/policy-controller/core/src/lib.rs
@@ -8,7 +8,7 @@ pub use self::{identity_match::IdentityMatch, network_match::NetworkMatch};
 use anyhow::Result;
 use futures::prelude::*;
 pub use ipnet::{IpNet, Ipv4Net, Ipv6Net};
-use std::{collections::HashMap, pin::Pin, time::Duration};
+use std::{collections::HashMap, hash::Hash, pin::Pin, time::Duration};
 
 /// Models inbound server configuration discovery.
 #[async_trait::async_trait]
@@ -28,7 +28,7 @@ pub struct InboundServer {
 }
 
 /// Describes how a proxy should handle inbound connections.
-#[derive(Clone, Debug, PartialEq, Eq)]
+#[derive(Clone, Debug, PartialEq, Eq, Hash)]
 pub enum ProxyProtocol {
     /// Indicates that the protocol should be discovered dynamically.
     Detect {
diff --git a/policy-controller/core/src/network_match.rs b/policy-controller/core/src/network_match.rs
index 506a29527..f87b32755 100644
--- a/policy-controller/core/src/network_match.rs
+++ b/policy-controller/core/src/network_match.rs
@@ -1,7 +1,7 @@
 use ipnet::{IpNet, Ipv4Net, Ipv6Net};
 use std::net::IpAddr;
 
-#[derive(Clone, Debug, PartialEq, Eq)]
+#[derive(Clone, Debug, PartialEq, Eq, Hash)]
 pub struct NetworkMatch {
     /// A network to match against.
     pub net: IpNet,
diff --git a/policy-controller/k8s/index/src/default_allow.rs b/policy-controller/k8s/index/src/default_allow.rs
deleted file mode 100644
index 5e89efb75..000000000
--- a/policy-controller/k8s/index/src/default_allow.rs
+++ /dev/null
@@ -1,204 +0,0 @@
-use crate::ServerRx;
-use anyhow::{anyhow, Error, Result};
-use linkerd_policy_controller_core::{
-    ClientAuthentication, ClientAuthorization, IdentityMatch, InboundServer, IpNet, NetworkMatch,
-    ProxyProtocol,
-};
-use linkerd_policy_controller_k8s_api as k8s;
-use std::future::Future;
-use tokio::{sync::watch, time};
-
-#[derive(Copy, Clone, Debug, PartialEq, Eq)]
-pub enum DefaultAllow {
-    AllAuthenticated,
-    AllUnauthenticated,
-    ClusterAuthenticated,
-    ClusterUnauthenticated,
-    Deny,
-}
-
-/// Default server configs to use when no server matches.
-#[derive(Clone, Debug)]
-pub(crate) struct DefaultAllowRxs {
-    all_authed_rx: ServerRx,
-    all_unauthed_rx: ServerRx,
-    cluster_authed_rx: ServerRx,
-    cluster_unauthed_rx: ServerRx,
-    deny_rx: ServerRx,
-}
-
-// === impl DefaultAllow ===
-
-impl DefaultAllow {
-    pub const ANNOTATION: &'static str = "policy.linkerd.io/default-allow";
-
-    pub fn from_annotation(meta: &k8s::ObjectMeta) -> Result<Option<Self>> {
-        if let Some(ann) = meta.annotations.as_ref() {
-            if let Some(v) = ann.get(Self::ANNOTATION) {
-                let mode = v.parse()?;
-                return Ok(Some(mode));
-            }
-        }
-
-        Ok(None)
-    }
-}
-
-impl std::str::FromStr for DefaultAllow {
-    type Err = Error;
-
-    fn from_str(s: &str) -> Result<Self> {
-        match s {
-            "all-authenticated" => Ok(Self::AllAuthenticated),
-            "all-unauthenticated" => Ok(Self::AllUnauthenticated),
-            "cluster-authenticated" => Ok(Self::ClusterAuthenticated),
-            "cluster-unauthenticated" => Ok(Self::ClusterUnauthenticated),
-            "deny" => Ok(Self::Deny),
-            s => Err(anyhow!("invalid mode: {:?}", s)),
-        }
-    }
-}
-
-impl std::fmt::Display for DefaultAllow {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        match self {
-            Self::AllAuthenticated => "all-authenticated".fmt(f),
-            Self::AllUnauthenticated => "all-unauthenticated".fmt(f),
-            Self::ClusterAuthenticated => "cluster-authenticated".fmt(f),
-            Self::ClusterUnauthenticated => "cluster-unauthenticated".fmt(f),
-            Self::Deny => "deny".fmt(f),
-        }
-    }
-}
-
-// === impl DefaultAllowRxs ===
-
-impl DefaultAllowRxs {
-    /// Create default allow policy receivers.
-    ///
-    /// These receivers are never updated. The senders are moved into a background task so that
-    /// the receivers continue to be live. The returned background task completes once all receivers
-    /// are dropped.
-    pub(crate) fn new(
-        cluster_nets: Vec<IpNet>,
-        detect_timeout: time::Duration,
-    ) -> (Self, impl Future<Output = ()> + Send) {
-        let any_authenticated =
-            ClientAuthentication::TlsAuthenticated(vec![IdentityMatch::Suffix(vec![])]);
-
-        let all_nets = [IpNet::V4(Default::default()), IpNet::V6(Default::default())];
-
-        let (all_authed_tx, all_authed_rx) = watch::channel(mk_detect_config(
-            "_all_authed",
-            detect_timeout,
-            all_nets.iter().cloned(),
-            any_authenticated.clone(),
-        ));
-
-        let (all_unauthed_tx, all_unauthed_rx) = watch::channel(mk_detect_config(
-            "_all_unauthed",
-            detect_timeout,
-            all_nets.iter().cloned(),
-            ClientAuthentication::Unauthenticated,
-        ));
-
-        let (cluster_authed_tx, cluster_authed_rx) = watch::channel(mk_detect_config(
-            "_cluster_authed",
-            detect_timeout,
-            cluster_nets.iter().cloned(),
-            any_authenticated,
-        ));
-
-        let (cluster_unauthed_tx, cluster_unauthed_rx) = watch::channel(mk_detect_config(
-            "_cluster_unauthed",
-            detect_timeout,
-            cluster_nets.into_iter(),
-            ClientAuthentication::Unauthenticated,
-        ));
-
-        let (deny_tx, deny_rx) = watch::channel(InboundServer {
-            protocol: ProxyProtocol::Detect {
-                timeout: detect_timeout,
-            },
-            authorizations: Default::default(),
-        });
-
-        // Ensure the senders are not dropped until all receivers are dropped.
-        let task = async move {
-            tokio::join!(
-                all_authed_tx.closed(),
-                all_unauthed_tx.closed(),
-                cluster_authed_tx.closed(),
-                cluster_unauthed_tx.closed(),
-                deny_tx.closed(),
-            );
-        };
-
-        let rxs = Self {
-            all_authed_rx,
-            all_unauthed_rx,
-            cluster_authed_rx,
-            cluster_unauthed_rx,
-            deny_rx,
-        };
-
-        (rxs, task)
-    }
-
-    pub fn get(&self, mode: DefaultAllow) -> ServerRx {
-        match mode {
-            DefaultAllow::AllAuthenticated => self.all_authed_rx.clone(),
-            DefaultAllow::AllUnauthenticated => self.all_unauthed_rx.clone(),
-            DefaultAllow::ClusterAuthenticated => self.cluster_authed_rx.clone(),
-            DefaultAllow::ClusterUnauthenticated => self.cluster_unauthed_rx.clone(),
-            DefaultAllow::Deny => self.deny_rx.clone(),
-        }
-    }
-}
-
-fn mk_detect_config(
-    name: &'static str,
-    timeout: time::Duration,
-    nets: impl IntoIterator<Item = IpNet>,
-    authentication: ClientAuthentication,
-) -> InboundServer {
-    let networks = nets
-        .into_iter()
-        .map(|net| NetworkMatch {
-            net,
-            except: vec![],
-        })
-        .collect::<Vec<_>>();
-    let authz = ClientAuthorization {
-        networks,
-        authentication,
-    };
-
-    InboundServer {
-        protocol: ProxyProtocol::Detect { timeout },
-        authorizations: Some((name.to_string(), authz)).into_iter().collect(),
-    }
-}
-
-#[cfg(test)]
-mod test {
-    use super::*;
-
-    #[test]
-    fn test_parse_displayed() {
-        for default in [
-            DefaultAllow::Deny,
-            DefaultAllow::AllAuthenticated,
-            DefaultAllow::AllUnauthenticated,
-            DefaultAllow::ClusterAuthenticated,
-            DefaultAllow::ClusterUnauthenticated,
-        ] {
-            assert_eq!(
-                default.to_string().parse::<DefaultAllow>().unwrap(),
-                default,
-                "failed to parse displayed {:?}",
-                default
-            );
-        }
-    }
-}
diff --git a/policy-controller/k8s/index/src/defaults.rs b/policy-controller/k8s/index/src/defaults.rs
new file mode 100644
index 000000000..01c3ef98c
--- /dev/null
+++ b/policy-controller/k8s/index/src/defaults.rs
@@ -0,0 +1,260 @@
+use crate::{ServerRx, ServerTx};
+use anyhow::{anyhow, Error, Result};
+use linkerd_policy_controller_core::{
+    ClientAuthentication, ClientAuthorization, IdentityMatch, InboundServer, IpNet, NetworkMatch,
+    ProxyProtocol,
+};
+use linkerd_policy_controller_k8s_api as k8s;
+use std::{collections::HashMap, hash::Hash};
+use tokio::{sync::watch, time};
+
+/// Indicates the default behavior to apply when no Server is found for a port.
+#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash)]
+pub enum DefaultPolicy {
+    Allow {
+        /// Indicates that, by default, all traffic must be authenticated.
+        authenticated_only: bool,
+
+        /// Indicates that all traffic must, by default, be from an IP address within the cluster.
+        cluster_only: bool,
+    },
+
+    /// Indicates that all traffic is denied unless explicitly permitted by an authorization policy.
+    Deny,
+}
+
+/// Describes the default behavior for a pod-port to apply when no Server is found for a port.
+#[derive(Copy, Clone, Debug, Default, PartialEq, Eq, Hash)]
+pub(crate) struct PortDefaults {
+    pub authenticated: bool,
+    pub opaque: bool,
+}
+
+/// Holds the watches for all default policies. These watches are initiated lazily and are never
+/// updated. `DefaultPolicyWatches` creates a watch as needed and holding its sender so that the
+/// receiver never observes the sender closing (or updating at all).
+#[derive(Debug)]
+pub(crate) struct DefaultPolicyWatches {
+    cluster_nets: Vec<IpNet>,
+    detect_timeout: time::Duration,
+
+    watches: HashMap<(DefaultPolicy, PortDefaults), (ServerTx, ServerRx)>,
+}
+
+// === impl DefaultPolicy ===
+
+impl DefaultPolicy {
+    pub const ANNOTATION: &'static str = "config.linkerd.io/default-inbound-policy";
+
+    pub fn from_annotation(meta: &k8s::ObjectMeta) -> Result<Option<Self>> {
+        if let Some(ann) = meta.annotations.as_ref() {
+            if let Some(v) = ann.get(Self::ANNOTATION) {
+                let mode = v.parse()?;
+                return Ok(Some(mode));
+            }
+        }
+
+        Ok(None)
+    }
+}
+
+impl std::str::FromStr for DefaultPolicy {
+    type Err = Error;
+
+    fn from_str(s: &str) -> Result<Self> {
+        match s {
+            "all-authenticated" => Ok(Self::Allow {
+                authenticated_only: true,
+                cluster_only: false,
+            }),
+            "all-unauthenticated" => Ok(Self::Allow {
+                authenticated_only: false,
+                cluster_only: false,
+            }),
+            "cluster-authenticated" => Ok(Self::Allow {
+                authenticated_only: true,
+                cluster_only: true,
+            }),
+            "cluster-unauthenticated" => Ok(Self::Allow {
+                authenticated_only: false,
+                cluster_only: true,
+            }),
+            "deny" => Ok(Self::Deny),
+            s => Err(anyhow!("invalid mode: {:?}", s)),
+        }
+    }
+}
+
+impl std::fmt::Display for DefaultPolicy {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::Allow {
+                authenticated_only: true,
+                cluster_only: false,
+            } => "all-authenticated".fmt(f),
+            Self::Allow {
+                authenticated_only: false,
+                cluster_only: false,
+            } => "all-unauthenticated".fmt(f),
+            Self::Allow {
+                authenticated_only: true,
+                cluster_only: true,
+            } => "cluster-authenticated".fmt(f),
+            Self::Allow {
+                authenticated_only: false,
+                cluster_only: true,
+            } => "cluster-unauthenticated".fmt(f),
+            Self::Deny => "deny".fmt(f),
+        }
+    }
+}
+
+// === impl DefaultPolicyWatches ===
+
+impl DefaultPolicyWatches {
+    /// Create default allow policy receivers.
+    ///
+    /// These receivers are never updated. The senders are moved into a background task so that
+    /// the receivers continue to be live. The returned background task completes once all receivers
+    /// are dropped.
+    pub(crate) fn new(cluster_nets: Vec<IpNet>, detect_timeout: time::Duration) -> Self {
+        Self {
+            cluster_nets,
+            detect_timeout,
+            watches: HashMap::default(),
+        }
+    }
+
+    /// Obtains a watch for a default policy. This watch never updates.
+    ///
+    /// If a watch for this policy does not already exist, one is created.
+    pub fn watch(&mut self, default: DefaultPolicy, config: PortDefaults) -> ServerRx {
+        use std::collections::hash_map::Entry;
+        match self.watches.entry((default, config)) {
+            Entry::Occupied(entry) => entry.get().1.clone(),
+            Entry::Vacant(entry) => {
+                let server =
+                    Self::mk_server(default, config, &self.cluster_nets, self.detect_timeout);
+                let (tx, rx) = watch::channel(server);
+                entry.insert((tx, rx.clone()));
+                rx
+            }
+        }
+    }
+
+    fn mk_server(
+        default: DefaultPolicy,
+        port: PortDefaults,
+        cluster_nets: &[IpNet],
+        detect_timeout: time::Duration,
+    ) -> InboundServer {
+        let protocol = Self::mk_protocol(detect_timeout, port);
+
+        match default {
+            DefaultPolicy::Allow {
+                cluster_only,
+                authenticated_only,
+            } => {
+                let name = Self::name(default, port);
+                let nets = if cluster_only {
+                    cluster_nets.to_vec()
+                } else {
+                    vec![IpNet::V4(Default::default()), IpNet::V6(Default::default())]
+                };
+                let authn = if authenticated_only || port.authenticated {
+                    let all_authed = IdentityMatch::Suffix(vec![]);
+                    ClientAuthentication::TlsAuthenticated(vec![all_authed])
+                } else {
+                    ClientAuthentication::Unauthenticated
+                };
+                Self::mk_policy(name, protocol, nets, authn)
+            }
+
+            DefaultPolicy::Deny => InboundServer {
+                protocol,
+                authorizations: Default::default(),
+            },
+        }
+    }
+
+    fn name(default: DefaultPolicy, port: PortDefaults) -> String {
+        // Update the default to account for whether the port requires authentication:
+        let default = match default {
+            DefaultPolicy::Deny => DefaultPolicy::Deny,
+            DefaultPolicy::Allow {
+                cluster_only,
+                authenticated_only,
+            } => DefaultPolicy::Allow {
+                cluster_only,
+                authenticated_only: authenticated_only || port.authenticated,
+            },
+        };
+        format!("default:{}", default)
+    }
+
+    fn mk_protocol(timeout: time::Duration, port: PortDefaults) -> ProxyProtocol {
+        if port.opaque {
+            return ProxyProtocol::Opaque;
+        }
+        ProxyProtocol::Detect { timeout }
+    }
+
+    fn mk_policy(
+        name: String,
+        protocol: ProxyProtocol,
+        nets: impl IntoIterator<Item = IpNet>,
+        authentication: ClientAuthentication,
+    ) -> InboundServer {
+        let networks = nets
+            .into_iter()
+            .map(|net| NetworkMatch {
+                net,
+                except: vec![],
+            })
+            .collect::<Vec<_>>();
+        let authz = ClientAuthorization {
+            networks,
+            authentication,
+        };
+
+        InboundServer {
+            protocol,
+            authorizations: Some((name, authz)).into_iter().collect(),
+        }
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+
+    #[test]
+    fn test_parse_displayed() {
+        for default in [
+            DefaultPolicy::Deny,
+            DefaultPolicy::Allow {
+                authenticated_only: true,
+                cluster_only: false,
+            },
+            DefaultPolicy::Allow {
+                authenticated_only: false,
+                cluster_only: false,
+            },
+            DefaultPolicy::Allow {
+                authenticated_only: false,
+                cluster_only: true,
+            },
+            DefaultPolicy::Allow {
+                authenticated_only: false,
+                cluster_only: true,
+            },
+        ] {
+            assert_eq!(
+                default.to_string().parse::<DefaultPolicy>().unwrap(),
+                default,
+                "failed to parse displayed {:?}",
+                default
+            );
+        }
+    }
+}
diff --git a/policy-controller/k8s/index/src/lib.rs b/policy-controller/k8s/index/src/lib.rs
index db26c7a44..0dfabb984 100644
--- a/policy-controller/k8s/index/src/lib.rs
+++ b/policy-controller/k8s/index/src/lib.rs
@@ -14,7 +14,7 @@
 //!   `ServerAuthorization` is updated, we find all of the `Server` instances it selects and update
 //!   their authorizations and publishes these updates on the server's broadcast channel.
 //!
-//! ```ignore
+//! ```text
 //! [Node] <- [ Pod ]
 //!           |-> [ Port ] <- [ Server ] <- [ ServerAuthorization ]
 //! ```
@@ -31,7 +31,7 @@
 #![forbid(unsafe_code)]
 
 mod authz;
-mod default_allow;
+mod defaults;
 mod lookup;
 mod namespace;
 mod node;
@@ -40,9 +40,9 @@ mod server;
 #[cfg(test)]
 mod tests;
 
-pub use self::{default_allow::DefaultAllow, lookup::Reader};
+pub use self::{defaults::DefaultPolicy, lookup::Reader};
 use self::{
-    default_allow::DefaultAllowRxs,
+    defaults::DefaultPolicyWatches,
     namespace::{Namespace, NamespaceIndex},
     node::NodeIndex,
     server::SrvIndex,
@@ -50,7 +50,6 @@ use self::{
 use anyhow::Context;
 use linkerd_policy_controller_core::{InboundServer, IpNet};
 use linkerd_policy_controller_k8s_api::{self as k8s, ResourceExt};
-use std::{future::Future, pin::Pin};
 use tokio::{sync::watch, time};
 use tracing::{debug, warn};
 
@@ -87,14 +86,10 @@ pub struct Index {
     /// Holds watches for the cluster's default-allow policies. These watches are never updated but
     /// this state is held so we can used shared references when updating a pod-port's server watch
     /// with a default policy.
-    default_allows: DefaultAllowRxs,
+    default_policy_watches: DefaultPolicyWatches,
 
     /// A handle that supports updates to the lookup index.
     lookups: lookup::Writer,
-
-    /// Holds the `DefaultAllowRxs` senders so that the receivers never signal an ending. This doesn't
-    /// actually need to be polled.
-    _default_allows_txs: Pin<Box<dyn Future<Output = ()> + Send + 'static>>,
 }
 
 #[derive(Debug)]
@@ -106,16 +101,16 @@ impl Index {
     pub fn new(
         cluster_networks: Vec<IpNet>,
         identity_domain: String,
-        default_allow: DefaultAllow,
+        default_policy: DefaultPolicy,
         detect_timeout: time::Duration,
     ) -> (lookup::Reader, Self) {
         // Create a common set of receivers for all supported default policies.
-        let (default_allows, _default_allows_txs) =
-            DefaultAllowRxs::new(cluster_networks.clone(), detect_timeout);
+        let default_policy_watches =
+            DefaultPolicyWatches::new(cluster_networks.clone(), detect_timeout);
 
         // Provide the cluster-wide default-allow policy to the namespace index so that it may be
         // used when a workload-level annotation is not set.
-        let namespaces = NamespaceIndex::new(default_allow);
+        let namespaces = NamespaceIndex::new(default_policy);
 
         let (writer, reader) = lookup::pair();
         let idx = Self {
@@ -123,9 +118,8 @@ impl Index {
             namespaces,
             identity_domain,
             cluster_networks,
-            default_allows,
+            default_policy_watches,
             nodes: NodeIndex::default(),
-            _default_allows_txs: Box::pin(_default_allows_txs),
         };
         (reader, idx)
     }
diff --git a/policy-controller/k8s/index/src/lookup.rs b/policy-controller/k8s/index/src/lookup.rs
index 317e1e359..ccd137f5c 100644
--- a/policy-controller/k8s/index/src/lookup.rs
+++ b/policy-controller/k8s/index/src/lookup.rs
@@ -185,7 +185,7 @@ impl Rx {
             authentication: ClientAuthentication::Unauthenticated,
         };
 
-        inner.authorizations.insert("_health_check".into(), authz);
+        inner.authorizations.insert("default:kubelet".into(), authz);
         inner
     }
 }
diff --git a/policy-controller/k8s/index/src/namespace.rs b/policy-controller/k8s/index/src/namespace.rs
index 7f0020edf..be1013ce1 100644
--- a/policy-controller/k8s/index/src/namespace.rs
+++ b/policy-controller/k8s/index/src/namespace.rs
@@ -1,4 +1,4 @@
-use crate::{authz::AuthzIndex, pod::PodIndex, server::SrvIndex, DefaultAllow};
+use crate::{authz::AuthzIndex, pod::PodIndex, server::SrvIndex, DefaultPolicy};
 use std::collections::HashMap;
 
 #[derive(Debug)]
@@ -6,13 +6,13 @@ pub(crate) struct NamespaceIndex {
     pub index: HashMap<String, Namespace>,
 
     // The global default-allow policy.
-    default_allow: DefaultAllow,
+    default_policy: DefaultPolicy,
 }
 
 #[derive(Debug)]
 pub(crate) struct Namespace {
     /// Holds the global default-allow policy, which may be overridden per-workload.
-    pub default_allow: DefaultAllow,
+    pub default_policy: DefaultPolicy,
 
     pub pods: PodIndex,
     pub servers: SrvIndex,
@@ -22,17 +22,17 @@ pub(crate) struct Namespace {
 // === impl Namespaces ===
 
 impl NamespaceIndex {
-    pub fn new(default_allow: DefaultAllow) -> Self {
+    pub fn new(default_policy: DefaultPolicy) -> Self {
         Self {
-            default_allow,
+            default_policy,
             index: HashMap::default(),
         }
     }
 
     pub fn get_or_default(&mut self, name: impl Into<String>) -> &mut Namespace {
-        let default_allow = self.default_allow;
+        let default_policy = self.default_policy;
         self.index.entry(name.into()).or_insert_with(|| Namespace {
-            default_allow,
+            default_policy,
             pods: PodIndex::default(),
             servers: SrvIndex::default(),
             authzs: AuthzIndex::default(),
diff --git a/policy-controller/k8s/index/src/pod.rs b/policy-controller/k8s/index/src/pod.rs
index 4e6a32f9d..037689f52 100644
--- a/policy-controller/k8s/index/src/pod.rs
+++ b/policy-controller/k8s/index/src/pod.rs
@@ -1,12 +1,12 @@
 use crate::{
-    lookup, node::KubeletIps, DefaultAllow, Errors, Index, Namespace, NodeIndex, PodServerTx,
-    ServerRx, SrvIndex,
+    defaults::PortDefaults, lookup, node::KubeletIps, DefaultPolicy, DefaultPolicyWatches, Errors,
+    Index, Namespace, NodeIndex, PodServerTx, ServerRx, SrvIndex,
 };
-use anyhow::{anyhow, Result};
+use anyhow::{anyhow, bail, Context, Result};
 use linkerd_policy_controller_k8s_api::{self as k8s, policy, ResourceExt};
 use std::collections::{hash_map::Entry as HashEntry, HashMap, HashSet};
 use tokio::sync::watch;
-use tracing::{debug, instrument, trace, warn};
+use tracing::{debug, instrument, trace};
 
 /// Indexes pod state (within a namespace).
 #[derive(Debug, Default)]
@@ -24,7 +24,7 @@ struct Pod {
     labels: k8s::Labels,
 
     /// The workload's default allow behavior (to apply when no `Server` references a port).
-    default_allow_rx: ServerRx,
+    default_policy: DefaultPolicy,
 }
 
 /// An index of all ports in a pod spec to the
@@ -41,9 +41,17 @@ struct PodPorts {
     by_name: HashMap<String, Vec<u16>>,
 }
 
+#[derive(Debug, Default)]
+struct PodAnnotations {
+    require_id: HashSet<u16>,
+    opaque: HashSet<u16>,
+}
+
 /// A single pod-port's state.
 #[derive(Debug)]
 struct Port {
+    default_policy_rx: ServerRx,
+
     /// Set with the name of the `Server` resource that currently selects this port, if one exists.
     ///
     /// When this is `None`, a default policy currently applies.
@@ -66,7 +74,7 @@ impl Index {
     )]
     pub(crate) fn apply_pod(&mut self, pod: k8s::Pod) -> Result<()> {
         let Namespace {
-            default_allow,
+            default_policy,
             ref mut pods,
             ref mut servers,
             ..
@@ -74,17 +82,13 @@ impl Index {
             .namespaces
             .get_or_default(pod.namespace().expect("namespace must be set"));
 
-        let default_allow = *default_allow;
-        let allows = self.default_allows.clone();
-        let mk_default_allow =
-            move |da: Option<DefaultAllow>| allows.get(da.unwrap_or(default_allow));
-
         pods.apply(
             pod,
             &mut self.nodes,
             servers,
             &mut self.lookups,
-            mk_default_allow,
+            *default_policy,
+            &mut self.default_policy_watches,
         )
     }
 
@@ -172,13 +176,12 @@ impl PodIndex {
 
     pub(crate) fn reset_server(&mut self, name: &str) {
         for (pod_name, pod) in self.index.iter_mut() {
-            let rx = pod.default_allow_rx.clone();
             for (p, port) in pod.ports.by_port.iter_mut() {
                 if port.server_name.as_deref() == Some(name) {
                     debug!(pod = %pod_name, port = %p, "Removing server from pod");
                     port.server_name = None;
                     port.server_tx
-                        .send(rx.clone())
+                        .send(port.default_policy_rx.clone())
                         .expect("pod config receiver must still be held");
                 } else {
                     trace!(pod = %pod_name, port = %p, server = ?port.server_name, "Server does not match");
@@ -194,7 +197,8 @@ impl PodIndex {
         nodes: &mut NodeIndex,
         servers: &SrvIndex,
         lookups: &mut lookup::Writer,
-        get_default_allow_rx: impl Fn(Option<DefaultAllow>) -> ServerRx,
+        default_policy: DefaultPolicy,
+        default_policy_watches: &mut DefaultPolicyWatches,
     ) -> Result<()> {
         let ns_name = pod.namespace().expect("pod must have a namespace");
         let pod_name = pod.name();
@@ -216,24 +220,31 @@ impl PodIndex {
                 // Check the pod for a default-allow annotation. If it's set, use it; otherwise use
                 // the default policy from the namespace or cluster. We retain this value (and not
                 // only the policy) so that we can more conveniently de-duplicate changes
-                let default_allow_rx = match DefaultAllow::from_annotation(&pod.metadata) {
-                    Ok(allow) => get_default_allow_rx(allow),
+                let default_policy = match DefaultPolicy::from_annotation(&pod.metadata) {
+                    Ok(allow) => allow.unwrap_or(default_policy),
                     Err(error) => {
-                        warn!(%error, "Ignoring invalid default-allow annotation");
-                        get_default_allow_rx(None)
+                        tracing::info!(%error, "failed to parse default-allow annotation");
+                        default_policy
                     }
                 };
 
+                let pod_annotations = PodAnnotations::from_annotations(&pod.metadata);
+
                 // Read the pod's ports and extract:
                 // - `ServerTx`s to be linkerd against the server index; and
                 // - lookup receivers to be returned to API clients.
-                let (ports, pod_lookups) =
-                    Self::extract_ports(spec, default_allow_rx.clone(), kubelet);
+                let (ports, pod_lookups) = Self::extract_ports(
+                    spec,
+                    &pod_annotations,
+                    default_policy,
+                    default_policy_watches,
+                    kubelet,
+                );
 
                 // Start tracking the pod's metadata so it can be linked against servers as they are
-                // created. Immediately link the pod against the server index.
+                // created. Immediately link the pod's ports to the server index.
                 let mut pod = Pod {
-                    default_allow_rx,
+                    default_policy,
                     labels: pod.metadata.labels.into(),
                     ports,
                 };
@@ -274,7 +285,9 @@ impl PodIndex {
     /// Extracts port information from a pod spec.
     fn extract_ports(
         spec: k8s::PodSpec,
-        server_rx: ServerRx,
+        annotations: &PodAnnotations,
+        default_policy: DefaultPolicy,
+        default_policy_watches: &mut DefaultPolicyWatches,
         kubelet: KubeletIps,
     ) -> (PodPorts, HashMap<u16, lookup::Rx>) {
         let mut ports = PodPorts::default();
@@ -289,8 +302,11 @@ impl PodIndex {
                         continue;
                     }
 
-                    let (server_tx, rx) = watch::channel(server_rx.clone());
+                    let config = annotations.get_config(port);
+                    let default_policy_rx = default_policy_watches.watch(default_policy, config);
+                    let (server_tx, rx) = watch::channel(default_policy_rx.clone());
                     let pod_port = Port {
+                        default_policy_rx,
                         server_name: None,
                         server_tx,
                     };
@@ -334,7 +350,7 @@ impl Pod {
             let port = self.ports.by_port.get_mut(&p).unwrap();
             port.server_name = None;
             port.server_tx
-                .send(self.default_allow_rx.clone())
+                .send(port.default_policy_rx.clone())
                 .expect("pod config receiver must still be held");
         }
     }
@@ -368,6 +384,80 @@ impl Pod {
     }
 }
 
+// === impl PodAnnotations ===
+
+impl PodAnnotations {
+    const OPAQUE_PORTS_ANNOTATION: &'static str = "config.linkerd.io/opaque-ports";
+    const REQUIRE_IDENTITY_PORTS_ANNOTATION: &'static str =
+        "config.linkerd.io/proxy-require-identity-inbound-ports";
+
+    fn from_annotations(meta: &k8s::ObjectMeta) -> Self {
+        let anns = match meta.annotations.as_ref() {
+            None => return Self::default(),
+            Some(anns) => anns,
+        };
+
+        let opaque = Self::ports_annotation(anns, Self::OPAQUE_PORTS_ANNOTATION);
+        let require_id = Self::ports_annotation(anns, Self::REQUIRE_IDENTITY_PORTS_ANNOTATION);
+
+        Self { opaque, require_id }
+    }
+
+    /// Reads `annotation` from the provided set of annotations, parsing it as a port set.  If the
+    /// annotation is not set or is invalid, the empty set is returned.
+    fn ports_annotation(
+        annotations: &std::collections::BTreeMap<String, String>,
+        annotation: &str,
+    ) -> HashSet<u16> {
+        annotations
+            .get(annotation)
+            .map(|spec| {
+                Self::parse_portset(spec).unwrap_or_else(|error| {
+                    tracing::info!(%spec, %error, %annotation, "Invalid ports list");
+                    Default::default()
+                })
+            })
+            .unwrap_or_default()
+    }
+
+    /// Read a comma-separated of ports or port ranges from the given string.
+    fn parse_portset(s: &str) -> Result<HashSet<u16>> {
+        let mut ports = HashSet::new();
+        for spec in s.split(',') {
+            match spec.split_once('-') {
+                None => {
+                    if !spec.trim().is_empty() {
+                        let port = spec.trim().parse().context("parsing port")?;
+                        if port == 0 {
+                            bail!("port must not be 0")
+                        }
+                        ports.insert(port);
+                    }
+                }
+                Some((floor, ceil)) => {
+                    let floor = floor.trim().parse::<u16>().context("parsing port")?;
+                    let ceil = ceil.trim().parse::<u16>().context("parsing port")?;
+                    if floor == 0 {
+                        bail!("port must not be 0")
+                    }
+                    if floor > ceil {
+                        bail!("Port range must be increasing");
+                    }
+                    ports.extend(floor..=ceil);
+                }
+            }
+        }
+        Ok(ports)
+    }
+
+    fn get_config(&self, port: u16) -> PortDefaults {
+        PortDefaults {
+            authenticated: self.require_id.contains(&port),
+            opaque: self.opaque.contains(&port),
+        }
+    }
+}
+
 // === impl PodPorts ===
 
 impl PodPorts {
@@ -384,3 +474,35 @@ impl PodPorts {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::PodAnnotations;
+
+    #[test]
+    fn parse_portset() {
+        assert!(
+            PodAnnotations::parse_portset("").unwrap().is_empty(),
+            "empty"
+        );
+        assert!(PodAnnotations::parse_portset("0").is_err(), "0");
+        assert_eq!(
+            PodAnnotations::parse_portset("1").unwrap(),
+            vec![1].into_iter().collect(),
+            "1"
+        );
+        assert_eq!(
+            PodAnnotations::parse_portset("1-2").unwrap(),
+            vec![1, 2].into_iter().collect(),
+            "1-2"
+        );
+        assert_eq!(
+            PodAnnotations::parse_portset("4,1-2").unwrap(),
+            vec![1, 2, 4].into_iter().collect(),
+            "4,1-2"
+        );
+        assert!(PodAnnotations::parse_portset("2-1").is_err(), "2-1");
+        assert!(PodAnnotations::parse_portset("2-").is_err(), "2-");
+        assert!(PodAnnotations::parse_portset("65537").is_err(), "65537");
+    }
+}
diff --git a/policy-controller/k8s/index/src/server.rs b/policy-controller/k8s/index/src/server.rs
index 71333ce1f..e483f214d 100644
--- a/policy-controller/k8s/index/src/server.rs
+++ b/policy-controller/k8s/index/src/server.rs
@@ -64,7 +64,7 @@ impl Index {
             ref mut pods,
             ref mut authzs,
             ref mut servers,
-            default_allow: _,
+            default_policy: _,
         } = self.namespaces.get_or_default(ns_name);
 
         servers.apply(srv, authzs);
diff --git a/policy-controller/src/main.rs b/policy-controller/src/main.rs
index 399020e65..bd72c9570 100644
--- a/policy-controller/src/main.rs
+++ b/policy-controller/src/main.rs
@@ -3,7 +3,7 @@
 
 use anyhow::{Context, Result};
 use futures::{future, prelude::*};
-use linkerd_policy_controller::k8s::DefaultAllow;
+use linkerd_policy_controller::k8s::DefaultPolicy;
 use linkerd_policy_controller_core::IpNet;
 use std::net::SocketAddr;
 use structopt::StructOpt;
@@ -32,7 +32,7 @@ struct Args {
     identity_domain: String,
 
     #[structopt(long, default_value = "all-unauthenticated")]
-    default_allow: DefaultAllow,
+    default_policy: DefaultPolicy,
 }
 
 #[tokio::main]
@@ -44,7 +44,7 @@ async fn main() -> Result<()> {
         grpc_addr,
         identity_domain,
         cluster_networks: IpNets(cluster_networks),
-        default_allow,
+        default_policy,
     } = Args::from_args();
 
     let (drain_tx, drain_rx) = drain::channel();
@@ -68,7 +68,7 @@ async fn main() -> Result<()> {
         let (handle, index) = linkerd_policy_controller::k8s::Index::new(
             cluster_networks.clone(),
             identity_domain,
-            default_allow,
+            default_policy,
             DETECT_TIMEOUT,
         );
 


Test patch:
diff --git a/cli/cmd/testdata/install_controlplane_tracing_output.golden b/cli/cmd/testdata/install_controlplane_tracing_output.golden
index b4dada36f..60dfea72c 100644
--- a/cli/cmd/testdata/install_controlplane_tracing_output.golden
+++ b/cli/cmd/testdata/install_controlplane_tracing_output.golden
@@ -1745,7 +1745,7 @@ spec:
         - --grpc-addr=0.0.0.0:8090
         - --cluster-networks=10.0.0.0/8,100.64.0.0/10,172.16.0.0/12,192.168.0.0/16
         - --identity-domain=cluster.local
-        - --default-allow=all-unauthenticated
+        - --default-policy=all-unauthenticated
         env:
         - name: RUST_LOG
           value: "linkerd=info,warn"
diff --git a/cli/cmd/testdata/install_custom_domain.golden b/cli/cmd/testdata/install_custom_domain.golden
index 70fdee547..2c4cb6a7e 100644
--- a/cli/cmd/testdata/install_custom_domain.golden
+++ b/cli/cmd/testdata/install_custom_domain.golden
@@ -1743,7 +1743,7 @@ spec:
         - --grpc-addr=0.0.0.0:8090
         - --cluster-networks=10.0.0.0/8,100.64.0.0/10,172.16.0.0/12,192.168.0.0/16
         - --identity-domain=cluster.local
-        - --default-allow=all-unauthenticated
+        - --default-policy=all-unauthenticated
         env:
         - name: RUST_LOG
           value: "linkerd=info,warn"
diff --git a/cli/cmd/testdata/install_custom_registry.golden b/cli/cmd/testdata/install_custom_registry.golden
index 247f0a0fa..ec55072e8 100644
--- a/cli/cmd/testdata/install_custom_registry.golden
+++ b/cli/cmd/testdata/install_custom_registry.golden
@@ -1743,7 +1743,7 @@ spec:
         - --grpc-addr=0.0.0.0:8090
         - --cluster-networks=10.0.0.0/8,100.64.0.0/10,172.16.0.0/12,192.168.0.0/16
         - --identity-domain=cluster.local
-        - --default-allow=all-unauthenticated
+        - --default-policy=all-unauthenticated
         env:
         - name: RUST_LOG
           value: "linkerd=info,warn"
diff --git a/cli/cmd/testdata/install_default.golden b/cli/cmd/testdata/install_default.golden
index 4d8032263..f320d43ae 100644
--- a/cli/cmd/testdata/install_default.golden
+++ b/cli/cmd/testdata/install_default.golden
@@ -1743,7 +1743,7 @@ spec:
         - --grpc-addr=0.0.0.0:8090
         - --cluster-networks=10.0.0.0/8,100.64.0.0/10,172.16.0.0/12,192.168.0.0/16
         - --identity-domain=cluster.local
-        - --default-allow=all-unauthenticated
+        - --default-policy=all-unauthenticated
         env:
         - name: RUST_LOG
           value: "linkerd=info,warn"
diff --git a/cli/cmd/testdata/install_default_override_dst_get_nets.golden b/cli/cmd/testdata/install_default_override_dst_get_nets.golden
index 24c1cc184..81e75df71 100644
--- a/cli/cmd/testdata/install_default_override_dst_get_nets.golden
+++ b/cli/cmd/testdata/install_default_override_dst_get_nets.golden
@@ -1743,7 +1743,7 @@ spec:
         - --grpc-addr=0.0.0.0:8090
         - --cluster-networks=10.0.0.0/8,100.64.0.0/10,172.0.0.0/8
         - --identity-domain=cluster.local
-        - --default-allow=all-unauthenticated
+        - --default-policy=all-unauthenticated
         env:
         - name: RUST_LOG
           value: "linkerd=info,warn"
diff --git a/cli/cmd/testdata/install_ha_output.golden b/cli/cmd/testdata/install_ha_output.golden
index d2cc39bd0..9766e2596 100644
--- a/cli/cmd/testdata/install_ha_output.golden
+++ b/cli/cmd/testdata/install_ha_output.golden
@@ -1867,7 +1867,7 @@ spec:
         - --grpc-addr=0.0.0.0:8090
         - --cluster-networks=10.0.0.0/8,100.64.0.0/10,172.16.0.0/12,192.168.0.0/16
         - --identity-domain=cluster.local
-        - --default-allow=all-unauthenticated
+        - --default-policy=all-unauthenticated
         env:
         - name: RUST_LOG
           value: "linkerd=info,warn"
diff --git a/cli/cmd/testdata/install_ha_with_overrides_output.golden b/cli/cmd/testdata/install_ha_with_overrides_output.golden
index d6f6c8e12..0ebe03caf 100644
--- a/cli/cmd/testdata/install_ha_with_overrides_output.golden
+++ b/cli/cmd/testdata/install_ha_with_overrides_output.golden
@@ -1867,7 +1867,7 @@ spec:
         - --grpc-addr=0.0.0.0:8090
         - --cluster-networks=10.0.0.0/8,100.64.0.0/10,172.16.0.0/12,192.168.0.0/16
         - --identity-domain=cluster.local
-        - --default-allow=all-unauthenticated
+        - --default-policy=all-unauthenticated
         env:
         - name: RUST_LOG
           value: "linkerd=info,warn"
diff --git a/cli/cmd/testdata/install_heartbeat_disabled_output.golden b/cli/cmd/testdata/install_heartbeat_disabled_output.golden
index faefce417..81231e711 100644
--- a/cli/cmd/testdata/install_heartbeat_disabled_output.golden
+++ b/cli/cmd/testdata/install_heartbeat_disabled_output.golden
@@ -1674,7 +1674,7 @@ spec:
         - --grpc-addr=0.0.0.0:8090
         - --cluster-networks=10.0.0.0/8,100.64.0.0/10,172.16.0.0/12,192.168.0.0/16
         - --identity-domain=cluster.local
-        - --default-allow=all-unauthenticated
+        - --default-policy=all-unauthenticated
         env:
         - name: RUST_LOG
           value: "linkerd=info,warn"
diff --git a/cli/cmd/testdata/install_helm_output.golden b/cli/cmd/testdata/install_helm_output.golden
index a3b11fda2..6ef46c040 100644
--- a/cli/cmd/testdata/install_helm_output.golden
+++ b/cli/cmd/testdata/install_helm_output.golden
@@ -1736,7 +1736,7 @@ spec:
         - --grpc-addr=0.0.0.0:8090
         - --cluster-networks=10.0.0.0/8,100.64.0.0/10,172.16.0.0/12,192.168.0.0/16
         - --identity-domain=test.trust.domain
-        - --default-allow=all-unauthenticated
+        - --default-policy=all-unauthenticated
         env:
         - name: RUST_LOG
           value: "linkerd=info,warn"
diff --git a/cli/cmd/testdata/install_helm_output_ha.golden b/cli/cmd/testdata/install_helm_output_ha.golden
index 45c0f22e3..67af0ec93 100644
--- a/cli/cmd/testdata/install_helm_output_ha.golden
+++ b/cli/cmd/testdata/install_helm_output_ha.golden
@@ -1860,7 +1860,7 @@ spec:
         - --grpc-addr=0.0.0.0:8090
         - --cluster-networks=10.0.0.0/8,100.64.0.0/10,172.16.0.0/12,192.168.0.0/16
         - --identity-domain=test.trust.domain
-        - --default-allow=all-unauthenticated
+        - --default-policy=all-unauthenticated
         env:
         - name: RUST_LOG
           value: "linkerd=info,warn"
diff --git a/cli/cmd/testdata/install_helm_output_ha_labels.golden b/cli/cmd/testdata/install_helm_output_ha_labels.golden
index e90b95979..e73fc2912 100644
--- a/cli/cmd/testdata/install_helm_output_ha_labels.golden
+++ b/cli/cmd/testdata/install_helm_output_ha_labels.golden
@@ -1872,7 +1872,7 @@ spec:
         - --grpc-addr=0.0.0.0:8090
         - --cluster-networks=10.0.0.0/8,100.64.0.0/10,172.16.0.0/12,192.168.0.0/16
         - --identity-domain=test.trust.domain
-        - --default-allow=all-unauthenticated
+        - --default-policy=all-unauthenticated
         env:
         - name: RUST_LOG
           value: "linkerd=info,warn"
diff --git a/cli/cmd/testdata/install_helm_output_ha_namespace_selector.golden b/cli/cmd/testdata/install_helm_output_ha_namespace_selector.golden
index af3fa2a6c..fbcbd6f55 100644
--- a/cli/cmd/testdata/install_helm_output_ha_namespace_selector.golden
+++ b/cli/cmd/testdata/install_helm_output_ha_namespace_selector.golden
@@ -1860,7 +1860,7 @@ spec:
         - --grpc-addr=0.0.0.0:8090
         - --cluster-networks=10.0.0.0/8,100.64.0.0/10,172.16.0.0/12,192.168.0.0/16
         - --identity-domain=test.trust.domain
-        - --default-allow=all-unauthenticated
+        - --default-policy=all-unauthenticated
         env:
         - name: RUST_LOG
           value: "linkerd=info,warn"
diff --git a/cli/cmd/testdata/install_no_init_container.golden b/cli/cmd/testdata/install_no_init_container.golden
index 4188d6451..bf71ab032 100644
--- a/cli/cmd/testdata/install_no_init_container.golden
+++ b/cli/cmd/testdata/install_no_init_container.golden
@@ -1705,7 +1705,7 @@ spec:
         - --grpc-addr=0.0.0.0:8090
         - --cluster-networks=10.0.0.0/8,100.64.0.0/10,172.16.0.0/12,192.168.0.0/16
         - --identity-domain=cluster.local
-        - --default-allow=all-unauthenticated
+        - --default-policy=all-unauthenticated
         env:
         - name: RUST_LOG
           value: "linkerd=info,warn"
diff --git a/cli/cmd/testdata/install_output.golden b/cli/cmd/testdata/install_output.golden
index d55b0e612..ef7895317 100644
--- a/cli/cmd/testdata/install_output.golden
+++ b/cli/cmd/testdata/install_output.golden
@@ -1747,7 +1747,7 @@ spec:
         - --grpc-addr=0.0.0.0:8090
         - --cluster-networks=ClusterNetworks
         - --identity-domain=cluster.local
-        - --default-allow=default-allow-policy
+        - --default-policy=default-allow-policy
         env:
         - name: RUST_LOG
           value: "log-level"
diff --git a/cli/cmd/testdata/install_proxy_ignores.golden b/cli/cmd/testdata/install_proxy_ignores.golden
index 65435272c..ad96fbbf9 100644
--- a/cli/cmd/testdata/install_proxy_ignores.golden
+++ b/cli/cmd/testdata/install_proxy_ignores.golden
@@ -1743,7 +1743,7 @@ spec:
         - --grpc-addr=0.0.0.0:8090
         - --cluster-networks=10.0.0.0/8,100.64.0.0/10,172.16.0.0/12,192.168.0.0/16
         - --identity-domain=cluster.local
-        - --default-allow=all-unauthenticated
+        - --default-policy=all-unauthenticated
         env:
         - name: RUST_LOG
           value: "linkerd=info,warn"
diff --git a/cli/cmd/testdata/install_values_file.golden b/cli/cmd/testdata/install_values_file.golden
index 957457528..e7dab459d 100644
--- a/cli/cmd/testdata/install_values_file.golden
+++ b/cli/cmd/testdata/install_values_file.golden
@@ -1729,7 +1729,7 @@ spec:
         - --grpc-addr=0.0.0.0:8090
         - --cluster-networks=10.0.0.0/8,100.64.0.0/10,172.16.0.0/12,192.168.0.0/16
         - --identity-domain=example.com
-        - --default-allow=all-unauthenticated
+        - --default-policy=all-unauthenticated
         env:
         - name: RUST_LOG
           value: "linkerd=info,warn"
diff --git a/policy-controller/k8s/index/src/tests.rs b/policy-controller/k8s/index/src/tests.rs
index 534f34cc9..a90f5e00c 100644
--- a/policy-controller/k8s/index/src/tests.rs
+++ b/policy-controller/k8s/index/src/tests.rs
@@ -22,7 +22,10 @@ async fn incrementally_configure_server() {
     let (lookup_rx, mut idx) = Index::new(
         vec![cluster_net],
         "cluster.example.com".into(),
-        DefaultAllow::ClusterUnauthenticated,
+        DefaultPolicy::Allow {
+            authenticated_only: false,
+            cluster_only: true,
+        },
         detect_timeout,
     );
 
@@ -38,8 +41,11 @@ async fn incrementally_configure_server() {
     idx.apply_pod(pod.clone()).unwrap();
 
     let default_config = InboundServer {
-        authorizations: mk_default_allow(
-            DefaultAllow::ClusterUnauthenticated,
+        authorizations: mk_default_policy(
+            DefaultPolicy::Allow {
+                authenticated_only: false,
+                cluster_only: true,
+            },
             cluster_net,
             kubelet_ip,
         ),
@@ -135,8 +141,8 @@ async fn incrementally_configure_server() {
     );
 }
 
-#[tokio::test]
-async fn server_update_deselects_pod() {
+#[test]
+fn server_update_deselects_pod() {
     let cluster_net = IpNet::from_str("192.0.2.0/24").unwrap();
     let pod_net = IpNet::from_str("192.0.2.2/28").unwrap();
     let (kubelet_ip, pod_ip) = {
@@ -147,7 +153,10 @@ async fn server_update_deselects_pod() {
     let (lookup_rx, mut idx) = Index::new(
         vec![cluster_net],
         "cluster.example.com".into(),
-        DefaultAllow::ClusterUnauthenticated,
+        DefaultPolicy::Allow {
+            authenticated_only: false,
+            cluster_only: true,
+        },
         detect_timeout,
     );
 
@@ -186,8 +195,11 @@ async fn server_update_deselects_pod() {
     assert_eq!(
         port2222.get(),
         InboundServer {
-            authorizations: mk_default_allow(
-                DefaultAllow::ClusterUnauthenticated,
+            authorizations: mk_default_policy(
+                DefaultPolicy::Allow {
+                    authenticated_only: false,
+                    cluster_only: true,
+                },
                 cluster_net,
                 kubelet_ip
             ),
@@ -198,11 +210,11 @@ async fn server_update_deselects_pod() {
     );
 }
 
-/// Tests that pod servers are configured with defaults based on the global `DefaultAllow` policy.
+/// Tests that pod servers are configured with defaults based on the global `DefaultPolicy` policy.
 ///
 /// Iterates through each default policy and validates that it produces expected configurations.
-#[tokio::test]
-async fn default_allow_global() {
+#[test]
+fn default_policy_global() {
     let cluster_net = IpNet::from_str("192.0.2.0/24").unwrap();
     let pod_net = IpNet::from_str("192.0.2.2/28").unwrap();
     let (kubelet_ip, pod_ip) = {
@@ -211,13 +223,7 @@ async fn default_allow_global() {
     };
     let detect_timeout = time::Duration::from_secs(1);
 
-    for default in &[
-        DefaultAllow::Deny,
-        DefaultAllow::AllAuthenticated,
-        DefaultAllow::AllUnauthenticated,
-        DefaultAllow::ClusterAuthenticated,
-        DefaultAllow::ClusterUnauthenticated,
-    ] {
+    for default in &DEFAULTS {
         let (lookup_rx, mut idx) = Index::new(
             vec![cluster_net],
             "cluster.example.com".into(),
@@ -237,7 +243,7 @@ async fn default_allow_global() {
         idx.reset_pods(vec![p]).unwrap();
 
         let config = InboundServer {
-            authorizations: mk_default_allow(*default, cluster_net, kubelet_ip),
+            authorizations: mk_default_policy(*default, cluster_net, kubelet_ip),
             protocol: ProxyProtocol::Detect {
                 timeout: detect_timeout,
             },
@@ -251,12 +257,12 @@ async fn default_allow_global() {
     }
 }
 
-/// Tests that pod servers are configured with defaults based on the workload-defined `DefaultAllow`
+/// Tests that pod servers are configured with defaults based on the workload-defined `DefaultPolicy`
 /// policy.
 ///
 /// Iterates through each default policy and validates that it produces expected configurations.
-#[tokio::test]
-async fn default_allow_annotated() {
+#[test]
+fn default_policy_annotated() {
     let cluster_net = IpNet::from_str("192.0.2.0/24").unwrap();
     let pod_net = IpNet::from_str("192.0.2.2/28").unwrap();
     let (kubelet_ip, pod_ip) = {
@@ -265,19 +271,17 @@ async fn default_allow_annotated() {
     };
     let detect_timeout = time::Duration::from_secs(1);
 
-    for default in &[
-        DefaultAllow::Deny,
-        DefaultAllow::AllAuthenticated,
-        DefaultAllow::AllUnauthenticated,
-        DefaultAllow::ClusterAuthenticated,
-        DefaultAllow::ClusterUnauthenticated,
-    ] {
+    for default in &DEFAULTS {
         let (lookup_rx, mut idx) = Index::new(
             vec![cluster_net],
             "cluster.example.com".into(),
+            // Invert default to ensure override applies.
             match *default {
-                DefaultAllow::Deny => DefaultAllow::AllUnauthenticated,
-                _ => DefaultAllow::Deny,
+                DefaultPolicy::Deny => DefaultPolicy::Allow {
+                    authenticated_only: false,
+                    cluster_only: false,
+                },
+                _ => DefaultPolicy::Deny,
             },
             detect_timeout,
         );
@@ -292,11 +296,11 @@ async fn default_allow_annotated() {
             Some(("container-0", vec![2222])),
         );
         p.annotations_mut()
-            .insert(DefaultAllow::ANNOTATION.into(), default.to_string());
+            .insert(DefaultPolicy::ANNOTATION.into(), default.to_string());
         idx.reset_pods(vec![p]).unwrap();
 
         let config = InboundServer {
-            authorizations: mk_default_allow(*default, cluster_net, kubelet_ip),
+            authorizations: mk_default_policy(*default, cluster_net, kubelet_ip),
             protocol: ProxyProtocol::Detect {
                 timeout: detect_timeout,
             },
@@ -308,10 +312,9 @@ async fn default_allow_annotated() {
         assert_eq!(port2222.get(), config);
     }
 }
-
 /// Tests that an invalid workload annotation is ignored in favor of the global default.
-#[tokio::test]
-async fn default_allow_annotated_invalid() {
+#[test]
+fn default_policy_annotated_invalid() {
     let cluster_net = IpNet::from_str("192.0.2.0/24").unwrap();
     let pod_net = IpNet::from_str("192.0.2.2/28").unwrap();
     let (kubelet_ip, pod_ip) = {
@@ -323,7 +326,10 @@ async fn default_allow_annotated_invalid() {
     let (lookup_rx, mut idx) = Index::new(
         vec![cluster_net],
         "cluster.example.com".into(),
-        DefaultAllow::AllUnauthenticated,
+        DefaultPolicy::Allow {
+            authenticated_only: false,
+            cluster_only: false,
+        },
         detect_timeout,
     );
 
@@ -337,7 +343,7 @@ async fn default_allow_annotated_invalid() {
         Some(("container-0", vec![2222])),
     );
     p.annotations_mut()
-        .insert(DefaultAllow::ANNOTATION.into(), "bogus".into());
+        .insert(DefaultPolicy::ANNOTATION.into(), "bogus".into());
     idx.reset_pods(vec![p]).unwrap();
 
     // Lookup port 2222 -> default config.
@@ -347,8 +353,11 @@ async fn default_allow_annotated_invalid() {
     assert_eq!(
         port2222.get(),
         InboundServer {
-            authorizations: mk_default_allow(
-                DefaultAllow::AllUnauthenticated,
+            authorizations: mk_default_policy(
+                DefaultPolicy::Allow {
+                    authenticated_only: false,
+                    cluster_only: false,
+                },
                 cluster_net,
                 kubelet_ip
             ),
@@ -359,9 +368,110 @@ async fn default_allow_annotated_invalid() {
     );
 }
 
+#[test]
+fn opaque_annotated() {
+    let cluster_net = IpNet::from_str("192.0.2.0/24").unwrap();
+    let pod_net = IpNet::from_str("192.0.2.2/28").unwrap();
+    let (kubelet_ip, pod_ip) = {
+        let mut ips = pod_net.hosts();
+        (ips.next().unwrap(), ips.next().unwrap())
+    };
+    let detect_timeout = time::Duration::from_secs(1);
+
+    for default in &DEFAULTS {
+        let (lookup_rx, mut idx) = Index::new(
+            vec![cluster_net],
+            "cluster.example.com".into(),
+            *default,
+            detect_timeout,
+        );
+
+        idx.reset_nodes(vec![mk_node("node-0", pod_net)]).unwrap();
+
+        let mut p = mk_pod(
+            "ns-0",
+            "pod-0",
+            "node-0",
+            pod_ip,
+            Some(("container-0", vec![2222])),
+        );
+        p.annotations_mut()
+            .insert("config.linkerd.io/opaque-ports".into(), "2222".into());
+        idx.reset_pods(vec![p]).unwrap();
+
+        let config = InboundServer {
+            authorizations: mk_default_policy(*default, cluster_net, kubelet_ip),
+            protocol: ProxyProtocol::Opaque,
+        };
+
+        let port2222 = lookup_rx
+            .lookup("ns-0", "pod-0", 2222)
+            .expect("pod must exist in lookups");
+        assert_eq!(port2222.get(), config);
+    }
+}
+
+#[test]
+fn authenticated_annotated() {
+    let cluster_net = IpNet::from_str("192.0.2.0/24").unwrap();
+    let pod_net = IpNet::from_str("192.0.2.2/28").unwrap();
+    let (kubelet_ip, pod_ip) = {
+        let mut ips = pod_net.hosts();
+        (ips.next().unwrap(), ips.next().unwrap())
+    };
+    let detect_timeout = time::Duration::from_secs(1);
+
+    for default in &DEFAULTS {
+        let (lookup_rx, mut idx) = Index::new(
+            vec![cluster_net],
+            "cluster.example.com".into(),
+            *default,
+            detect_timeout,
+        );
+
+        idx.reset_nodes(vec![mk_node("node-0", pod_net)]).unwrap();
+
+        let mut p = mk_pod(
+            "ns-0",
+            "pod-0",
+            "node-0",
+            pod_ip,
+            Some(("container-0", vec![2222])),
+        );
+        p.annotations_mut().insert(
+            "config.linkerd.io/proxy-require-identity-inbound-ports".into(),
+            "2222".into(),
+        );
+        idx.reset_pods(vec![p]).unwrap();
+
+        let config = InboundServer {
+            authorizations: mk_default_policy(
+                match *default {
+                    DefaultPolicy::Allow { cluster_only, .. } => DefaultPolicy::Allow {
+                        cluster_only,
+                        authenticated_only: true,
+                    },
+                    DefaultPolicy::Deny => DefaultPolicy::Deny,
+                },
+                cluster_net,
+                kubelet_ip,
+            ),
+            protocol: ProxyProtocol::Detect {
+                timeout: detect_timeout,
+            },
+        };
+
+        let port2222 = lookup_rx
+            .lookup("ns-0", "pod-0", 2222)
+            .expect("pod must exist in lookups");
+        assert_eq!(port2222.get().protocol, config.protocol);
+        assert_eq!(port2222.get().authorizations, config.authorizations);
+    }
+}
+
 /// Tests observing a pod before its node has been observed amid resets.
-#[tokio::test]
-async fn pod_before_node_reset() {
+#[test]
+fn pod_before_node_reset() {
     let cluster_net = IpNet::from_str("192.0.2.0/24").unwrap();
     let pod_net = IpNet::from_str("192.0.2.2/28").unwrap();
     let (_kubelet_ip, pod_ip) = {
@@ -373,7 +483,7 @@ async fn pod_before_node_reset() {
     let (lookup_rx, mut idx) = Index::new(
         vec![cluster_net],
         "cluster.example.com".into(),
-        DefaultAllow::Deny,
+        DefaultPolicy::Deny,
         detect_timeout,
     );
 
@@ -413,8 +523,8 @@ async fn pod_before_node_reset() {
 }
 
 /// Tests observing a pod before its node has been observed amid resets.
-#[tokio::test]
-async fn pod_before_node_remove() {
+#[test]
+fn pod_before_node_remove() {
     let cluster_net = IpNet::from_str("192.0.2.0/24").unwrap();
     let pod_net = IpNet::from_str("192.0.2.2/28").unwrap();
     let (_kubelet_ip, pod_ip) = {
@@ -426,7 +536,7 @@ async fn pod_before_node_remove() {
     let (lookup_rx, mut idx) = Index::new(
         vec![cluster_net],
         "cluster.example.com".into(),
-        DefaultAllow::Deny,
+        DefaultPolicy::Deny,
         detect_timeout,
     );
 
@@ -454,6 +564,26 @@ async fn pod_before_node_remove() {
 
 // === Helpers ===
 
+const DEFAULTS: [DefaultPolicy; 5] = [
+    DefaultPolicy::Deny,
+    DefaultPolicy::Allow {
+        authenticated_only: true,
+        cluster_only: false,
+    },
+    DefaultPolicy::Allow {
+        authenticated_only: false,
+        cluster_only: false,
+    },
+    DefaultPolicy::Allow {
+        authenticated_only: true,
+        cluster_only: true,
+    },
+    DefaultPolicy::Allow {
+        authenticated_only: false,
+        cluster_only: true,
+    },
+];
+
 fn mk_node(name: impl Into<String>, pod_net: IpNet) -> k8s::Node {
     k8s::Node {
         metadata: k8s::ObjectMeta {
@@ -564,8 +694,8 @@ fn mk_authz(
     }
 }
 
-fn mk_default_allow(
-    da: DefaultAllow,
+fn mk_default_policy(
+    da: DefaultPolicy,
     cluster_net: IpNet,
     kubelet_ip: IpAddr,
 ) -> HashMap<String, ClientAuthorization> {
@@ -576,30 +706,42 @@ fn mk_default_allow(
     let authed = ClientAuthentication::TlsAuthenticated(vec![IdentityMatch::Suffix(vec![])]);
 
     match da {
-        DefaultAllow::Deny => None,
-        DefaultAllow::AllAuthenticated => Some((
-            "_all_authed".into(),
+        DefaultPolicy::Deny => None,
+        DefaultPolicy::Allow {
+            authenticated_only: true,
+            cluster_only: false,
+        } => Some((
+            "default:all-authenticated".into(),
             ClientAuthorization {
                 authentication: authed,
                 networks: all_nets,
             },
         )),
-        DefaultAllow::AllUnauthenticated => Some((
-            "_all_unauthed".into(),
+        DefaultPolicy::Allow {
+            authenticated_only: false,
+            cluster_only: false,
+        } => Some((
+            "default:all-unauthenticated".into(),
             ClientAuthorization {
                 authentication: ClientAuthentication::Unauthenticated,
                 networks: all_nets,
             },
         )),
-        DefaultAllow::ClusterAuthenticated => Some((
-            "_cluster_authed".into(),
+        DefaultPolicy::Allow {
+            authenticated_only: true,
+            cluster_only: true,
+        } => Some((
+            "default:cluster-authenticated".into(),
             ClientAuthorization {
                 authentication: authed,
                 networks: cluster_nets,
             },
         )),
-        DefaultAllow::ClusterUnauthenticated => Some((
-            "_cluster_unauthed".into(),
+        DefaultPolicy::Allow {
+            authenticated_only: false,
+            cluster_only: true,
+        } => Some((
+            "default:cluster-unauthenticated".into(),
             ClientAuthorization {
                 authentication: ClientAuthentication::Unauthenticated,
                 networks: cluster_nets,
@@ -613,7 +755,7 @@ fn mk_default_allow(
 
 fn healthcheck_authz(ip: IpAddr) -> (String, ClientAuthorization) {
     (
-        "_health_check".into(),
+        "default:kubelet".into(),
         ClientAuthorization {
             networks: vec![NetworkMatch::from(IpNet::from(ip))],
             authentication: ClientAuthentication::Unauthenticated,