diff --git a/cmd/machine-config-operator/start.go b/cmd/machine-config-operator/start.go
index bdf9a5adb5..b9ea06f953 100644
--- a/cmd/machine-config-operator/start.go
+++ b/cmd/machine-config-operator/start.go
@@ -120,6 +120,7 @@ func runStartCmd(_ *cobra.Command, _ []string) {
 			ctrlctx.ConfigInformerFactory.Config().V1().ClusterVersions(),
 			ctrlctx.InformerFactory.Machineconfiguration().V1().OSImageStreams(),
 			iriInformer,
+			ctrlctx.KubeNamespacedInformerFactory.Networking().V1().NetworkPolicies(),
 			ctrlctx,
 		)
 
diff --git a/manifests/machineconfigcontroller/networkpolicy-deletion-guard-validatingadmissionpolicy.yaml b/manifests/machineconfigcontroller/networkpolicy-deletion-guard-validatingadmissionpolicy.yaml
new file mode 100644
index 0000000000..78845af7de
--- /dev/null
+++ b/manifests/machineconfigcontroller/networkpolicy-deletion-guard-validatingadmissionpolicy.yaml
@@ -0,0 +1,21 @@
+apiVersion: admissionregistration.k8s.io/v1
+kind: ValidatingAdmissionPolicy
+metadata:
+  name: "networkpolicy-prevent-deletion"
+spec:
+  failurePolicy: Fail
+  matchConstraints:
+    matchPolicy: Equivalent
+    namespaceSelector:
+      matchLabels:
+        kubernetes.io/metadata.name: openshift-machine-config-operator
+    objectSelector: {}
+    resourceRules:
+    - apiGroups:   ["networking.k8s.io"]
+      apiVersions: ["v1"]
+      operations:  ["DELETE"]
+      resources:   ["networkpolicies"]
+      scope: "Namespaced"
+  validations:
+    - expression: "!(oldObject.metadata.name in ['default-deny', 'allow-machine-config-operator', 'allow-machine-config-controller', 'allow-machine-os-builder'])"
+      message: "Deletion of MCO-managed NetworkPolicy resources in the openshift-machine-config-operator namespace is not allowed"
diff --git a/manifests/machineconfigcontroller/networkpolicy-deletion-guard-validatingadmissionpolicybinding.yaml b/manifests/machineconfigcontroller/networkpolicy-deletion-guard-validatingadmissionpolicybinding.yaml
new file mode 100644
index 0000000000..81975ebe45
--- /dev/null
+++ b/manifests/machineconfigcontroller/networkpolicy-deletion-guard-validatingadmissionpolicybinding.yaml
@@ -0,0 +1,7 @@
+apiVersion: admissionregistration.k8s.io/v1
+kind: ValidatingAdmissionPolicyBinding
+metadata:
+  name: "networkpolicy-prevent-deletion-binding"
+spec:
+  policyName: "networkpolicy-prevent-deletion"
+  validationActions: [Deny]
diff --git a/pkg/operator/network_policy.go b/pkg/operator/network_policy.go
new file mode 100644
index 0000000000..c149fe67fb
--- /dev/null
+++ b/pkg/operator/network_policy.go
@@ -0,0 +1,99 @@
+package operator
+
+import (
+	"context"
+
+	configv1 "github.com/openshift/api/config/v1"
+	ctrlcommon "github.com/openshift/machine-config-operator/pkg/controller/common"
+	corev1 "k8s.io/api/core/v1"
+	networkingv1 "k8s.io/api/networking/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/intstr"
+	metav1ac "k8s.io/client-go/applyconfigurations/meta/v1"
+	networkingv1ac "k8s.io/client-go/applyconfigurations/networking/v1"
+)
+
+const networkPolicyFieldManager = "machine-config-operator"
+
+func defaultDenyNetworkPolicy(namespace string) *networkingv1ac.NetworkPolicyApplyConfiguration {
+	return networkingv1ac.NetworkPolicy("default-deny", namespace).
+		WithSpec(networkingv1ac.NetworkPolicySpec().
+			WithPodSelector(metav1ac.LabelSelector()).
+			WithPolicyTypes(networkingv1.PolicyTypeIngress, networkingv1.PolicyTypeEgress))
+}
+
+func allowMCONetworkPolicy(namespace string) *networkingv1ac.NetworkPolicyApplyConfiguration {
+	return networkingv1ac.NetworkPolicy("allow-machine-config-operator", namespace).
+		WithSpec(networkingv1ac.NetworkPolicySpec().
+			WithPodSelector(metav1ac.LabelSelector().
+				WithMatchLabels(map[string]string{"k8s-app": "machine-config-operator"})).
+			WithIngress(networkingv1ac.NetworkPolicyIngressRule().
+				WithPorts(networkingv1ac.NetworkPolicyPort().
+					WithProtocol(corev1.ProtocolTCP).
+					WithPort(intstr.FromInt32(9001)))).
+			WithEgress(networkingv1ac.NetworkPolicyEgressRule()).
+			WithPolicyTypes(networkingv1.PolicyTypeIngress, networkingv1.PolicyTypeEgress))
+}
+
+func allowMCCNetworkPolicy(namespace string) *networkingv1ac.NetworkPolicyApplyConfiguration {
+	return networkingv1ac.NetworkPolicy("allow-machine-config-controller", namespace).
+		WithSpec(networkingv1ac.NetworkPolicySpec().
+			WithPodSelector(metav1ac.LabelSelector().
+				WithMatchLabels(map[string]string{"k8s-app": "machine-config-controller"})).
+			WithIngress(networkingv1ac.NetworkPolicyIngressRule().
+				WithPorts(networkingv1ac.NetworkPolicyPort().
+					WithProtocol(corev1.ProtocolTCP).
+					WithPort(intstr.FromInt32(9001)))).
+			WithEgress(networkingv1ac.NetworkPolicyEgressRule()).
+			WithPolicyTypes(networkingv1.PolicyTypeIngress, networkingv1.PolicyTypeEgress))
+}
+
+func allowMOBNetworkPolicy(namespace string) *networkingv1ac.NetworkPolicyApplyConfiguration {
+	return networkingv1ac.NetworkPolicy("allow-machine-os-builder", namespace).
+		WithSpec(networkingv1ac.NetworkPolicySpec().
+			WithPodSelector(metav1ac.LabelSelector().
+				WithMatchLabels(map[string]string{"k8s-app": "machine-os-builder"})).
+			WithIngress(networkingv1ac.NetworkPolicyIngressRule().
+				WithPorts(networkingv1ac.NetworkPolicyPort().
+					WithProtocol(corev1.ProtocolTCP).
+					WithPort(intstr.FromInt32(9001)))).
+			WithEgress(networkingv1ac.NetworkPolicyEgressRule()).
+			WithPolicyTypes(networkingv1.PolicyTypeIngress, networkingv1.PolicyTypeEgress))
+}
+
+func (optr *Operator) syncNetworkPolicies(_ *renderConfig, _ *configv1.ClusterOperator) error {
+	ctx := context.TODO()
+	ns := ctrlcommon.MCONamespace
+	applyOpts := metav1.ApplyOptions{FieldManager: networkPolicyFieldManager, Force: true}
+
+	staticPolicies := []*networkingv1ac.NetworkPolicyApplyConfiguration{
+		defaultDenyNetworkPolicy(ns),
+		allowMCONetworkPolicy(ns),
+		allowMCCNetworkPolicy(ns),
+	}
+
+	for _, policy := range staticPolicies {
+		if _, err := optr.kubeClient.NetworkingV1().NetworkPolicies(ns).Apply(ctx, policy, applyOpts); err != nil {
+			return err
+		}
+	}
+
+	layeredMCPs, err := optr.getLayeredMachineConfigPools()
+	if err != nil {
+		return err
+	}
+
+	if len(layeredMCPs) > 0 {
+		if _, err := optr.kubeClient.NetworkingV1().NetworkPolicies(ns).Apply(ctx, allowMOBNetworkPolicy(ns), applyOpts); err != nil {
+			return err
+		}
+	} else {
+		err := optr.kubeClient.NetworkingV1().NetworkPolicies(ns).Delete(ctx, "allow-machine-os-builder", metav1.DeleteOptions{})
+		if err != nil && !apierrors.IsNotFound(err) {
+			return err
+		}
+	}
+
+	return nil
+}
diff --git a/pkg/operator/network_policy_test.go b/pkg/operator/network_policy_test.go
new file mode 100644
index 0000000000..cd9b05049d
--- /dev/null
+++ b/pkg/operator/network_policy_test.go
@@ -0,0 +1,190 @@
+package operator
+
+import (
+	"context"
+	"testing"
+
+	mcfgv1 "github.com/openshift/api/machineconfiguration/v1"
+	fakeclientmachineconfigv1 "github.com/openshift/client-go/machineconfiguration/clientset/versioned/fake"
+	mcfginformers "github.com/openshift/client-go/machineconfiguration/informers/externalversions"
+	"github.com/openshift/library-go/pkg/operator/events"
+	ctrlcommon "github.com/openshift/machine-config-operator/pkg/controller/common"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	networkingv1 "k8s.io/api/networking/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes/fake"
+	"k8s.io/utils/clock"
+)
+
+func testRenderConfig() *renderConfig {
+	return &renderConfig{
+		TargetNamespace: ctrlcommon.MCONamespace,
+	}
+}
+
+func testOperatorForNetworkPolicies(t *testing.T, pools []*mcfgv1.MachineConfigPool) *Operator {
+	t.Helper()
+
+	kubeClient := fake.NewClientset()
+	mcfgClient := fakeclientmachineconfigv1.NewSimpleClientset()
+	mcfgInformerFactory := mcfginformers.NewSharedInformerFactory(mcfgClient, 0)
+	mcpInformer := mcfgInformerFactory.Machineconfiguration().V1().MachineConfigPools()
+	moscInformer := mcfgInformerFactory.Machineconfiguration().V1().MachineOSConfigs()
+
+	for _, pool := range pools {
+		require.NoError(t, mcpInformer.Informer().GetIndexer().Add(pool))
+	}
+
+	return &Operator{
+		kubeClient:    kubeClient,
+		mcpLister:     mcpInformer.Lister(),
+		moscLister:    moscInformer.Lister(),
+		libgoRecorder: events.NewInMemoryRecorder("test-operator", clock.RealClock{}),
+	}
+}
+
+func layeredPool(name string) *mcfgv1.MachineConfigPool {
+	return &mcfgv1.MachineConfigPool{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: name,
+			Labels: map[string]string{
+				ctrlcommon.LayeringEnabledPoolLabel: "",
+			},
+		},
+	}
+}
+
+func listNetworkPolicies(t *testing.T, optr *Operator) []networkingv1.NetworkPolicy {
+	t.Helper()
+	npList, err := optr.kubeClient.NetworkingV1().NetworkPolicies(ctrlcommon.MCONamespace).List(context.TODO(), metav1.ListOptions{})
+	require.NoError(t, err)
+	return npList.Items
+}
+
+func findPolicy(policies []networkingv1.NetworkPolicy, name string) *networkingv1.NetworkPolicy {
+	for i := range policies {
+		if policies[i].Name == name {
+			return &policies[i]
+		}
+	}
+	return nil
+}
+
+func TestSyncNetworkPolicies_StaticPoliciesCreated(t *testing.T) {
+	optr := testOperatorForNetworkPolicies(t, nil)
+	config := testRenderConfig()
+
+	err := optr.syncNetworkPolicies(config, nil)
+	require.NoError(t, err)
+
+	policies := listNetworkPolicies(t, optr)
+	names := make([]string, len(policies))
+	for i, p := range policies {
+		names[i] = p.Name
+	}
+
+	assert.Contains(t, names, "default-deny", "expected default-deny policy")
+	assert.Contains(t, names, "allow-machine-config-operator", "expected allow-machine-config-operator policy")
+	assert.Contains(t, names, "allow-machine-config-controller", "expected allow-machine-config-controller policy")
+	assert.NotContains(t, names, "allow-machine-os-builder", "MOB policy should not be created without layered pools")
+}
+
+func TestSyncNetworkPolicies_DefaultDenySpec(t *testing.T) {
+	optr := testOperatorForNetworkPolicies(t, nil)
+	config := testRenderConfig()
+
+	err := optr.syncNetworkPolicies(config, nil)
+	require.NoError(t, err)
+
+	policies := listNetworkPolicies(t, optr)
+	denyPolicy := findPolicy(policies, "default-deny")
+	require.NotNil(t, denyPolicy, "default-deny policy must exist")
+
+	assert.Empty(t, denyPolicy.Spec.PodSelector.MatchLabels, "default-deny should select all pods with empty selector")
+	assert.Contains(t, denyPolicy.Spec.PolicyTypes, networkingv1.PolicyTypeIngress, "default-deny must include Ingress policy type")
+	assert.Contains(t, denyPolicy.Spec.PolicyTypes, networkingv1.PolicyTypeEgress, "default-deny must include Egress policy type")
+	assert.Empty(t, denyPolicy.Spec.Ingress, "default-deny should have no ingress rules")
+	assert.Empty(t, denyPolicy.Spec.Egress, "default-deny should have no egress rules")
+}
+
+func TestSyncNetworkPolicies_AllowPolicySpecs(t *testing.T) {
+	optr := testOperatorForNetworkPolicies(t, nil)
+	config := testRenderConfig()
+
+	err := optr.syncNetworkPolicies(config, nil)
+	require.NoError(t, err)
+
+	policies := listNetworkPolicies(t, optr)
+
+	cases := []struct {
+		name      string
+		labelKey  string
+		labelVal  string
+		metricsPort int32
+	}{
+		{"allow-machine-config-operator", "k8s-app", "machine-config-operator", 9001},
+		{"allow-machine-config-controller", "k8s-app", "machine-config-controller", 9001},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			policy := findPolicy(policies, tc.name)
+			require.NotNil(t, policy, "policy %s must exist", tc.name)
+
+			assert.Equal(t, tc.labelVal, policy.Spec.PodSelector.MatchLabels[tc.labelKey],
+				"podSelector should match %s=%s", tc.labelKey, tc.labelVal)
+			assert.Contains(t, policy.Spec.PolicyTypes, networkingv1.PolicyTypeIngress)
+			assert.Contains(t, policy.Spec.PolicyTypes, networkingv1.PolicyTypeEgress)
+
+			require.Len(t, policy.Spec.Egress, 1, "should have one egress rule (allow all)")
+			assert.Empty(t, policy.Spec.Egress[0].Ports, "egress rule should allow all ports")
+			assert.Empty(t, policy.Spec.Egress[0].To, "egress rule should allow all destinations")
+
+			require.Len(t, policy.Spec.Ingress, 1, "should have one ingress rule")
+			require.Len(t, policy.Spec.Ingress[0].Ports, 1, "ingress rule should have one port")
+			assert.Equal(t, tc.metricsPort, policy.Spec.Ingress[0].Ports[0].Port.IntVal, "ingress port should be metrics port")
+			assert.Empty(t, policy.Spec.Ingress[0].From, "ingress should not restrict source (kube-rbac-proxy handles auth)")
+		})
+	}
+}
+
+func TestSyncNetworkPolicies_MOBPolicyWithLayeredPools(t *testing.T) {
+	pools := []*mcfgv1.MachineConfigPool{layeredPool("layered-worker")}
+	optr := testOperatorForNetworkPolicies(t, pools)
+	config := testRenderConfig()
+
+	err := optr.syncNetworkPolicies(config, nil)
+	require.NoError(t, err)
+
+	policies := listNetworkPolicies(t, optr)
+	mobPolicy := findPolicy(policies, "allow-machine-os-builder")
+	require.NotNil(t, mobPolicy, "MOB policy should be created when layered pools exist")
+
+	assert.Equal(t, "machine-os-builder", mobPolicy.Spec.PodSelector.MatchLabels["k8s-app"])
+	require.Len(t, mobPolicy.Spec.Egress, 1, "MOB should have one egress rule (allow all)")
+	require.Len(t, mobPolicy.Spec.Ingress, 1, "MOB should have one ingress rule")
+}
+
+func TestSyncNetworkPolicies_MOBPolicyDeletedWithoutLayeredPools(t *testing.T) {
+	pools := []*mcfgv1.MachineConfigPool{layeredPool("layered-worker")}
+	optr := testOperatorForNetworkPolicies(t, pools)
+	config := testRenderConfig()
+
+	err := optr.syncNetworkPolicies(config, nil)
+	require.NoError(t, err)
+
+	policies := listNetworkPolicies(t, optr)
+	require.NotNil(t, findPolicy(policies, "allow-machine-os-builder"), "MOB policy should exist initially")
+
+	optr.mcpLister = testOperatorForNetworkPolicies(t, nil).mcpLister
+
+	err = optr.syncNetworkPolicies(config, nil)
+	require.NoError(t, err)
+
+	policies = listNetworkPolicies(t, optr)
+	assert.Nil(t, findPolicy(policies, "allow-machine-os-builder"), "MOB policy should be deleted when no layered pools exist")
+	assert.NotNil(t, findPolicy(policies, "default-deny"), "static policies should still exist")
+	assert.NotNil(t, findPolicy(policies, "allow-machine-config-operator"), "static policies should still exist")
+	assert.NotNil(t, findPolicy(policies, "allow-machine-config-controller"), "static policies should still exist")
+}
+
diff --git a/pkg/operator/operator.go b/pkg/operator/operator.go
index d512cbd8d3..389538a107 100644
--- a/pkg/operator/operator.go
+++ b/pkg/operator/operator.go
@@ -26,6 +26,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/wait"
 	appsinformersv1 "k8s.io/client-go/informers/apps/v1"
 	coreinformersv1 "k8s.io/client-go/informers/core/v1"
+	networkinginformersv1 "k8s.io/client-go/informers/networking/v1"
 	rbacinformersv1 "k8s.io/client-go/informers/rbac/v1"
 	"k8s.io/client-go/kubernetes"
 	coreclientsetv1 "k8s.io/client-go/kubernetes/typed/core/v1"
@@ -168,6 +169,7 @@ type Operator struct {
 	apiserverListerSynced            cache.InformerSynced
 	osImageStreamListerSynced        cache.InformerSynced
 	iriListerSynced                  cache.InformerSynced
+	networkPolicyInformerSynced      cache.InformerSynced
 
 	// queue only ever has one item, but it has nice error handling backoff/retry semantics
 	queue workqueue.TypedRateLimitingInterface[string]
@@ -229,6 +231,7 @@ func New(
 	clusterVersionInformer configinformersv1.ClusterVersionInformer,
 	osImageStreamInformer mcfginformersv1.OSImageStreamInformer,
 	iriInformer mcfginformersv1alpha1.InternalReleaseImageInformer,
+	networkPolicyInformer networkinginformersv1.NetworkPolicyInformer,
 	ctrlctx *ctrlcommon.ControllerContext,
 ) *Operator {
 	eventBroadcaster := record.NewBroadcaster()
@@ -301,6 +304,7 @@ func New(
 		clusterOperatorInformer.Informer(),
 		apiserverInformer.Informer(),
 		moscInformer.Informer(),
+		networkPolicyInformer.Informer(),
 	}
 	for _, i := range informers {
 		i.AddEventHandler(optr.eventHandler())
@@ -377,6 +381,7 @@ func New(
 	optr.apiserverListerSynced = apiserverInformer.Informer().HasSynced
 	optr.moscLister = moscInformer.Lister()
 	optr.moscListerSynced = moscInformer.Informer().HasSynced
+	optr.networkPolicyInformerSynced = networkPolicyInformer.Informer().HasSynced
 	optr.clusterVersionLister = clusterVersionInformer.Lister()
 	if osImageStreamInformer != nil && osimagestream.IsFeatureEnabled(optr.fgHandler) {
 		optr.osImageStreamLister = osImageStreamInformer.Lister()
@@ -464,6 +469,7 @@ func (optr *Operator) Run(workers int, stopCh <-chan struct{}) {
 		optr.crcListerSynced,
 		optr.nodeClusterListerSynced,
 		optr.moscListerSynced,
+		optr.networkPolicyInformerSynced,
 	}
 	if optr.osImageStreamListerSynced != nil && osimagestream.IsFeatureEnabled(optr.fgHandler) {
 		cacheSynced = append(cacheSynced, optr.osImageStreamListerSynced)
@@ -618,6 +624,7 @@ func (optr *Operator) sync(key string) error {
 		{"MachineConfiguration", optr.syncMachineConfiguration},
 		{"MachineConfigNode", optr.syncMachineConfigNodes},
 		{"MachineConfigPools", optr.syncMachineConfigPools},
+		{"NetworkPolicies", optr.syncNetworkPolicies},
 		{"MachineConfigDaemon", optr.syncMachineConfigDaemon},
 		{"MachineConfigController", optr.syncMachineConfigController},
 		{"MachineConfigServer", optr.syncMachineConfigServer},
diff --git a/pkg/operator/sync.go b/pkg/operator/sync.go
index 5eeda92e0c..ffa9b46420 100644
--- a/pkg/operator/sync.go
+++ b/pkg/operator/sync.go
@@ -113,6 +113,8 @@ const (
 	mccOSImageStreamDeletionGuardValidatingAdmissionPolicyBindingPath     = "manifests/machineconfigcontroller/osimagestream-deletion-guard-validatingadmissionpolicybinding.yaml"
 	mccIRIDeletionGuardValidatingAdmissionPolicyPath                      = "manifests/machineconfigcontroller/internalreleaseimage-deletion-guard-validatingadmissionpolicy.yaml"
 	mccIRIDeletionGuardValidatingAdmissionPolicyBindingPath               = "manifests/machineconfigcontroller/internalreleaseimage-deletion-guard-validatingadmissionpolicybinding.yaml"
+	mccNetworkPolicyDeletionGuardValidatingAdmissionPolicyPath            = "manifests/machineconfigcontroller/networkpolicy-deletion-guard-validatingadmissionpolicy.yaml"
+	mccNetworkPolicyDeletionGuardValidatingAdmissionPolicyBindingPath     = "manifests/machineconfigcontroller/networkpolicy-deletion-guard-validatingadmissionpolicybinding.yaml"
 	mccUpdateBootImagesCPMSValidatingAdmissionPolicyPath                  = "manifests/machineconfigcontroller/update-bootimages-cpms-validatingadmissionpolicy.yaml"
 	mccUpdateBootImagesCPMSValidatingAdmissionPolicyBindingPath           = "manifests/machineconfigcontroller/update-bootimages-cpms-validatingadmissionpolicybinding.yaml"
 
@@ -1210,11 +1212,13 @@ func (optr *Operator) syncMachineConfigController(config *renderConfig, _ *confi
 			mccMachineConfigurationGuardsValidatingAdmissionPolicyPath,
 			mccUpdateBootImagesValidatingAdmissionPolicyPath,
 			mccMachineConfigPoolSelectorValidatingAdmissionPolicyPath,
+			mccNetworkPolicyDeletionGuardValidatingAdmissionPolicyPath,
 		},
 		validatingAdmissionPolicyBindings: []string{
 			mccMachineConfigurationGuardsValidatingAdmissionPolicyBindingPath,
 			mccUpdateBootImagesValidatingAdmissionPolicyBindingPath,
 			mccMachineConfigPoolSelectorValidatingAdmissionPolicyBindingPath,
+			mccNetworkPolicyDeletionGuardValidatingAdmissionPolicyBindingPath,
 		},
 	}
 
diff --git a/test/e2e-2of2/network_policy_test.go b/test/e2e-2of2/network_policy_test.go
new file mode 100644
index 0000000000..02c54c25f1
--- /dev/null
+++ b/test/e2e-2of2/network_policy_test.go
@@ -0,0 +1,300 @@
+package e2e_2of2_test
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	configv1 "github.com/openshift/api/config/v1"
+	corev1 "k8s.io/api/core/v1"
+	networkingv1 "k8s.io/api/networking/v1"
+	k8serrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/intstr"
+	"k8s.io/apimachinery/pkg/util/wait"
+
+	ctrlcommon "github.com/openshift/machine-config-operator/pkg/controller/common"
+	"github.com/openshift/machine-config-operator/test/framework"
+)
+
+const (
+	networkPolicySyncPollInterval = 5 * time.Second
+	networkPolicySyncPollTimeout  = 2 * time.Minute
+)
+
+var staticPolicyNames = []string{
+	"default-deny",
+	"allow-machine-config-operator",
+	"allow-machine-config-controller",
+}
+
+func skipIfNoNetworkPolicies(t *testing.T) {
+	t.Helper()
+	cs := framework.NewClientSet("")
+	policies, err := cs.GetKubeclient().NetworkingV1().NetworkPolicies(ctrlcommon.MCONamespace).List(
+		context.Background(), metav1.ListOptions{})
+	require.NoError(t, err)
+	require.NotEmpty(t, policies.Items, "expected at least one NetworkPolicy in MCO namespace")
+}
+
+func TestNetworkPolicies_DefaultPoliciesExist(t *testing.T) {
+	skipIfNoNetworkPolicies(t)
+
+	cs := framework.NewClientSet("")
+	ctx := context.Background()
+	ns := ctrlcommon.MCONamespace
+
+	policies, err := cs.GetKubeclient().NetworkingV1().NetworkPolicies(ns).List(ctx, metav1.ListOptions{})
+	require.NoError(t, err)
+
+	policyMap := make(map[string]*networkingv1.NetworkPolicy)
+	for i := range policies.Items {
+		policyMap[policies.Items[i].Name] = &policies.Items[i]
+	}
+
+	for _, name := range staticPolicyNames {
+		assert.Contains(t, policyMap, name, "expected policy %s to exist", name)
+	}
+
+	deny := policyMap["default-deny"]
+	if deny != nil {
+		assert.Empty(t, deny.Spec.PodSelector.MatchLabels, "default-deny should select all pods")
+		assert.Contains(t, deny.Spec.PolicyTypes, networkingv1.PolicyTypeIngress)
+		assert.Contains(t, deny.Spec.PolicyTypes, networkingv1.PolicyTypeEgress)
+		assert.Empty(t, deny.Spec.Ingress, "default-deny should have no ingress rules")
+		assert.Empty(t, deny.Spec.Egress, "default-deny should have no egress rules")
+	}
+
+	for _, tc := range []struct {
+		name     string
+		labelVal string
+	}{
+		{"allow-machine-config-operator", "machine-config-operator"},
+		{"allow-machine-config-controller", "machine-config-controller"},
+	} {
+		policy := policyMap[tc.name]
+		if policy == nil {
+			continue
+		}
+
+		assert.Equal(t, tc.labelVal, policy.Spec.PodSelector.MatchLabels["k8s-app"],
+			"%s: podSelector should match k8s-app=%s", tc.name, tc.labelVal)
+		assert.Contains(t, policy.Spec.PolicyTypes, networkingv1.PolicyTypeIngress)
+		assert.Contains(t, policy.Spec.PolicyTypes, networkingv1.PolicyTypeEgress)
+
+		require.NotEmpty(t, policy.Spec.Ingress, "%s: should have ingress rules", tc.name)
+		require.NotEmpty(t, policy.Spec.Ingress[0].Ports, "%s: ingress should have ports", tc.name)
+		assert.Equal(t, int32(9001), policy.Spec.Ingress[0].Ports[0].Port.IntVal,
+			"%s: ingress port should be 9001", tc.name)
+		assert.Equal(t, corev1.ProtocolTCP, *policy.Spec.Ingress[0].Ports[0].Protocol,
+			"%s: ingress protocol should be TCP", tc.name)
+
+		require.NotEmpty(t, policy.Spec.Egress, "%s: should have egress rules", tc.name)
+	}
+}
+
+func TestNetworkPolicies_ModificationReverted(t *testing.T) {
+	skipIfNoNetworkPolicies(t)
+
+	cs := framework.NewClientSet("")
+	ctx := context.Background()
+	ns := ctrlcommon.MCONamespace
+	npClient := cs.GetKubeclient().NetworkingV1().NetworkPolicies(ns)
+
+	policy, err := npClient.Get(ctx, "allow-machine-config-operator", metav1.GetOptions{})
+	require.NoError(t, err)
+	require.NotEmpty(t, policy.Spec.Ingress, "policy should have ingress rules before modification")
+
+	policy.Spec.Ingress = nil
+	_, err = npClient.Update(ctx, policy, metav1.UpdateOptions{})
+	require.NoError(t, err)
+
+	modified, err := npClient.Get(ctx, "allow-machine-config-operator", metav1.GetOptions{})
+	require.NoError(t, err)
+	assert.Empty(t, modified.Spec.Ingress, "ingress should be empty after modification")
+
+	err = wait.PollUntilContextTimeout(ctx, networkPolicySyncPollInterval, networkPolicySyncPollTimeout, true,
+		func(ctx context.Context) (bool, error) {
+			restored, err := npClient.Get(ctx, "allow-machine-config-operator", metav1.GetOptions{})
+			if err != nil {
+				return false, err
+			}
+			return len(restored.Spec.Ingress) > 0, nil
+		})
+	require.NoError(t, err, "operator should revert modified ingress rules")
+
+	restored, err := npClient.Get(ctx, "allow-machine-config-operator", metav1.GetOptions{})
+	require.NoError(t, err)
+	require.Len(t, restored.Spec.Ingress, 1)
+	require.Len(t, restored.Spec.Ingress[0].Ports, 1)
+	assert.Equal(t, int32(9001), restored.Spec.Ingress[0].Ports[0].Port.IntVal,
+		"restored policy should have metrics port 9001")
+}
+
+func TestNetworkPolicies_DeletionBlocked(t *testing.T) {
+	skipIfNoNetworkPolicies(t)
+
+	cs := framework.NewClientSet("")
+	ctx := context.Background()
+	ns := ctrlcommon.MCONamespace
+	npClient := cs.GetKubeclient().NetworkingV1().NetworkPolicies(ns)
+
+	for _, name := range staticPolicyNames {
+		err := npClient.Delete(ctx, name, metav1.DeleteOptions{})
+		require.Error(t, err, "deleting managed policy %s should be blocked", name)
+		require.True(t, k8serrors.IsInvalid(err),
+			"error should be Invalid (422) from ValidatingAdmissionPolicy, got: %v", err)
+		require.Contains(t, err.Error(), "MCO-managed NetworkPolicy",
+			"error message should indicate the policy is MCO-managed")
+	}
+
+	// Verify all policies still exist after blocked deletion attempts
+	policies, err := npClient.List(ctx, metav1.ListOptions{})
+	require.NoError(t, err)
+
+	policyNames := make(map[string]bool)
+	for _, p := range policies.Items {
+		policyNames[p.Name] = true
+	}
+	for _, expected := range staticPolicyNames {
+		assert.True(t, policyNames[expected], "policy %s should still exist after blocked deletion", expected)
+	}
+}
+
+func TestNetworkPolicies_MCOProcessesContinueWorking(t *testing.T) {
+	skipIfNoNetworkPolicies(t)
+
+	cs := framework.NewClientSet("")
+	ctx := context.Background()
+	ns := ctrlcommon.MCONamespace
+
+	co, err := cs.ClusterOperators().Get(ctx, "machine-config", metav1.GetOptions{})
+	require.NoError(t, err, "should be able to get machine-config ClusterOperator")
+
+	foundAvailable, foundDegraded := false, false
+	for _, cond := range co.Status.Conditions {
+		switch cond.Type {
+		case configv1.OperatorAvailable:
+			foundAvailable = true
+			assert.Equal(t, configv1.ConditionTrue, cond.Status,
+				"machine-config operator should be Available")
+		case configv1.OperatorDegraded:
+			foundDegraded = true
+			assert.Equal(t, configv1.ConditionFalse, cond.Status,
+				"machine-config operator should not be Degraded")
+		}
+	}
+	assert.True(t, foundAvailable, "OperatorAvailable condition must be present on machine-config ClusterOperator")
+	assert.True(t, foundDegraded, "OperatorDegraded condition must be present on machine-config ClusterOperator")
+
+	pods, err := cs.Pods(ns).List(ctx, metav1.ListOptions{})
+	require.NoError(t, err)
+
+	componentRunning := map[string]bool{
+		"machine-config-operator":   false,
+		"machine-config-controller": false,
+	}
+	for _, pod := range pods.Items {
+		app := pod.Labels["k8s-app"]
+		if _, tracked := componentRunning[app]; tracked && pod.Status.Phase == corev1.PodRunning {
+			componentRunning[app] = true
+		}
+	}
+	for component, running := range componentRunning {
+		assert.True(t, running, "%s pod should be Running", component)
+	}
+
+	policies, err := cs.GetKubeclient().NetworkingV1().NetworkPolicies(ns).List(ctx, metav1.ListOptions{})
+	require.NoError(t, err)
+
+	policyNames := make([]string, len(policies.Items))
+	for i, p := range policies.Items {
+		policyNames[i] = p.Name
+	}
+	for _, expected := range staticPolicyNames {
+		assert.Contains(t, policyNames, expected,
+			"policy %s should exist while MCO processes are running", expected)
+	}
+}
+
+func TestNetworkPolicies_AdminNetworkPolicyOverride(t *testing.T) {
+	skipIfNoNetworkPolicies(t)
+
+	cs := framework.NewClientSet("")
+	ctx := context.Background()
+	ns := ctrlcommon.MCONamespace
+
+	deny, err := cs.GetKubeclient().NetworkingV1().NetworkPolicies(ns).Get(ctx, "default-deny", metav1.GetOptions{})
+	require.NoError(t, err)
+	assert.Empty(t, deny.Spec.Ingress, "default-deny should block all ingress")
+
+	overridePolicy := &networkingv1.NetworkPolicy{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "test-additional-allow",
+			Namespace: ns,
+		},
+		Spec: networkingv1.NetworkPolicySpec{
+			PodSelector: metav1.LabelSelector{
+				MatchLabels: map[string]string{"k8s-app": "machine-config-operator"},
+			},
+			Ingress: []networkingv1.NetworkPolicyIngressRule{
+				{
+					Ports: []networkingv1.NetworkPolicyPort{
+						{
+							Protocol: npProtocolPtr(corev1.ProtocolTCP),
+							Port:     npPortPtr(8080),
+						},
+					},
+				},
+			},
+			PolicyTypes: []networkingv1.PolicyType{networkingv1.PolicyTypeIngress},
+		},
+	}
+
+	_, err = cs.GetKubeclient().NetworkingV1().NetworkPolicies(ns).Create(ctx, overridePolicy, metav1.CreateOptions{})
+	require.NoError(t, err)
+	defer func() {
+		_ = cs.GetKubeclient().NetworkingV1().NetworkPolicies(ns).Delete(ctx, "test-additional-allow", metav1.DeleteOptions{})
+	}()
+
+	created, err := cs.GetKubeclient().NetworkingV1().NetworkPolicies(ns).Get(ctx, "test-additional-allow", metav1.GetOptions{})
+	require.NoError(t, err)
+	assert.Equal(t, "test-additional-allow", created.Name)
+
+	err = wait.PollUntilContextTimeout(ctx, networkPolicySyncPollInterval, networkPolicySyncPollTimeout, true,
+		func(ctx context.Context) (bool, error) {
+			policies, err := cs.GetKubeclient().NetworkingV1().NetworkPolicies(ns).List(ctx, metav1.ListOptions{})
+			if err != nil {
+				return false, err
+			}
+			names := make(map[string]bool)
+			for _, p := range policies.Items {
+				names[p.Name] = true
+			}
+			for _, expected := range staticPolicyNames {
+				if !names[expected] {
+					return false, nil
+				}
+			}
+			return names["test-additional-allow"], nil
+		})
+	require.NoError(t, err, "operator-managed policies should coexist with user-created policies")
+
+	final, err := cs.GetKubeclient().NetworkingV1().NetworkPolicies(ns).Get(ctx, "test-additional-allow", metav1.GetOptions{})
+	require.NoError(t, err)
+	require.Len(t, final.Spec.Ingress, 1)
+	require.Len(t, final.Spec.Ingress[0].Ports, 1)
+	assert.Equal(t, int32(8080), final.Spec.Ingress[0].Ports[0].Port.IntVal,
+		"user-created policy should retain its original port")
+}
+
+func npProtocolPtr(p corev1.Protocol) *corev1.Protocol {
+	return &p
+}
+
+func npPortPtr(port int32) *intstr.IntOrString {
+	p := intstr.FromInt32(port)
+	return &p
+}