Read Network Segments

We can create provider networks, but we also need to be able to read
them back out again, so OpenStack is the source of truth, rather than
having to cache information, which can go wrong!

Author: spjmurray

README.md

@@ -93,7 +93,7 @@ pip3 install build pylint
 
 ```bash
 python3 -m build
-pip3 install --force-reinstall dist/python_unikorn_openstack_policy-0.1.0-py3-none-any.whl
+pip3 install --force-reinstall --no-deps dist/python_unikorn_openstack_policy-0.1.0-py3-none-any.whl
 ```
 
 ### Generating Policy Files

pyproject.toml

@@ -23,11 +23,14 @@ classifiers = [
     "Programming Language :: Python :: 3.11",
     "Programming Language :: Python :: 3 :: Only",
 ]
+
+# Note this is based on Dalmation.
+# See: https://releases.openstack.org/
 dependencies = [
-	"cinder",
-	"neutron",
-	"nova",
-	"oslo.config",
+	"cinder ~= 25.0",
+	"neutron ~= 25.0",
+	"nova ~= 30.0",
+	"oslo.config ~= 9.5.0",
 ]
 
 [project.urls]

unikorn_openstack_policy/blockstorage.py

@@ -35,6 +35,7 @@
 ]
 
 
+# pylint: disable=R0801
 def list_rules():
     """Implements the "oslo.policy.policies" entry point"""
 
@@ -47,7 +48,10 @@ def list_rules():
 def get_enforcer():
     """Implements the "oslo.policy.enforcer" entry point"""
 
-    enforcer = policy.Enforcer(conf=cfg.CONF)
+    conf=cfg.CONF
+    conf(args=[])
+
+    enforcer = policy.Enforcer(conf=conf)
     enforcer.register_defaults(list_rules())
 
     return enforcer

unikorn_openstack_policy/compute.py

@@ -34,6 +34,7 @@
 ]
 
 
+# pylint: disable=R0801
 def list_rules():
     """Implements the "oslo.policy.policies" entry point"""
 
@@ -46,7 +47,10 @@ def list_rules():
 def get_enforcer():
     """Implements the "oslo.policy.enforcer" entry point"""
 
-    enforcer = policy.Enforcer(conf=cfg.CONF)
+    conf=cfg.CONF
+    conf(args=[])
+
+    enforcer = policy.Enforcer(conf=conf)
     enforcer.register_defaults(list_rules())
 
     return enforcer

unikorn_openstack_policy/network.py

@@ -58,6 +58,26 @@
         check_str='rule:is_project_manager',
         description='Specify ``provider:segmentation_id`` when creating a network',
     ),
+    policy.RuleDefault(
+        name='get_network:segments',
+        check_str='rule:is_project_manager',
+        description='Get ``segments`` attribute of a network',
+    ),
+    policy.RuleDefault(
+        name='get_network:provider:network_type',
+        check_str='rule:is_project_manager',
+        description='Get ``provider:network_type`` attribute of a network',
+    ),
+    policy.RuleDefault(
+        name='get_network:provider:physical_network',
+        check_str='rule:is_project_manager',
+        description='Get ``provider:physical_network`` attribute of a network',
+    ),
+    policy.RuleDefault(
+        name='get_network:provider:segmentation_id',
+        check_str='rule:is_project_manager',
+        description='Get ``provider:segmentation_id`` aattribute of a network',
+    ),
 
     # The domain manager can update quotas.
     policy.RuleDefault(
@@ -68,6 +88,7 @@
 ]
 
 
+# pylint: disable=R0801
 def list_rules():
     """Implements the "oslo.policy.policies" entry point"""
 
@@ -80,7 +101,10 @@ def list_rules():
 def get_enforcer():
     """Implements the "oslo.policy.enforcer" entry point"""
 
-    enforcer = policy.Enforcer(conf=cfg.CONF)
+    conf=cfg.CONF
+    conf(args=[])
+
+    enforcer = policy.Enforcer(conf=conf)
     enforcer.register_defaults(list_rules())
 
     return enforcer

unikorn_openstack_policy/tests/test_network.py

@@ -67,6 +67,34 @@ def test_create_network_provider_segmentation_id(self):
                 self.enforce(
                     'create_network:provider:segmentation_id', self.alt_target, self.context))
 
+    def test_get_network_segments(self):
+        """Admin can get network segments"""
+        self.assertTrue(self.enforce('get_network:segments', self.target, self.context))
+        self.assertTrue(self.enforce('get_network:segments', self.alt_target, self.context))
+
+    def test_get_network_provider_network_type(self):
+        """Admin can get provider network types"""
+        self.assertTrue(
+                self.enforce('create_network:provider:network_type', self.target, self.context))
+        self.assertTrue(
+                self.enforce('create_network:provider:network_type', self.alt_target, self.context))
+
+    def test_get_network_provider_physical_network(self):
+        """Admin can get provider physical networks"""
+        self.assertTrue(
+                self.enforce('create_network:provider:physical_network', self.target, self.context))
+        self.assertTrue(
+                self.enforce(
+                    'create_network:provider:physical_network', self.alt_target, self.context))
+
+    def test_get_network_provider_segmentation_id(self):
+        """Admin can get provider segmentation IDs"""
+        self.assertTrue(
+                self.enforce('create_network:provider:segmentation_id', self.target, self.context))
+        self.assertTrue(
+                self.enforce(
+                    'create_network:provider:segmentation_id', self.alt_target, self.context))
+
     def test_delete_network(self):
         """Admin can delete networks"""
         self.assertTrue(self.enforce('delete_network', self.target, self.context))
@@ -144,6 +172,40 @@ def test_create_network_provider_segmentation_id(self):
                 self.enforce,
                 'create_network:provider:segmentation_id', self.alt_target, self.context)
 
+    def test_get_network_segments(self):
+        """Admin can get network segments"""
+        self.assertTrue(self.enforce('get_network:segments', self.target, self.context))
+        self.assertRaises(
+                policy.PolicyNotAuthorized,
+                self.enforce, 'get_network:segments', self.alt_target, self.context)
+
+    def test_get_network_provider_network_type(self):
+        """Admin can get provider network types"""
+        self.assertTrue(
+                self.enforce('create_network:provider:network_type', self.target, self.context))
+        self.assertRaises(
+                policy.PolicyNotAuthorized,
+                self.enforce,
+                'create_network:provider:network_type', self.alt_target, self.context)
+
+    def test_get_network_provider_physical_network(self):
+        """Admin can get provider physical networks"""
+        self.assertTrue(
+                self.enforce('create_network:provider:physical_network', self.target, self.context))
+        self.assertRaises(
+                policy.PolicyNotAuthorized,
+                self.enforce,
+                'create_network:provider:physical_network', self.alt_target, self.context)
+
+    def test_get_network_provider_segmentation_id(self):
+        """Admin can get provider segmentation IDs"""
+        self.assertTrue(
+                self.enforce('create_network:provider:segmentation_id', self.target, self.context))
+        self.assertRaises(
+                policy.PolicyNotAuthorized,
+                self.enforce,
+                'create_network:provider:segmentation_id', self.alt_target, self.context)
+
     def test_delete_network(self):
         """Project manager can create networks"""
         self.assertTrue(self.enforce('delete_network', self.target, self.context))