diff --git a/CHANGELOG.md b/CHANGELOG.md
index 673537c..3b0a70d 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,6 +2,12 @@
 
 * Support added for config env variable expansion on nested levels, not just
   top-level provider/processor keys
+* `Processors.process_zone_config` method added to allow processors that work
+  with the zone config data. Configured with `manager.zone-processors: []`,
+  default is ['dynamic-zone-config']
+* Converted dynamic zone config to be a processors ^, if zone-processors are
+  explicitely configured and dynamic zone config is desired
+  `dyanmic-zone-config` must be included in the list as the desired position
 
 ## v1.4.0 - 2023-12-04 - Minor Meta
 
diff --git a/octodns/manager.py b/octodns/manager.py
index 0e76132..ee66acb 100644
--- a/octodns/manager.py
+++ b/octodns/manager.py
@@ -15,6 +15,7 @@ from . import __version__
 from .idna import IdnaDict, idna_decode, idna_encode
 from .processor.arpa import AutoArpa
 from .processor.meta import MetaProcessor
+from .processor.zone import DynamicZoneConfigProcessor
 from .provider.base import BaseProvider
 from .provider.plan import Plan
 from .provider.yaml import SplitYamlProvider, YamlProvider
@@ -111,6 +112,11 @@ class Manager(object):
 
         self.auto_arpa = self._config_auto_arpa(manager_config, auto_arpa)
 
+        self.zone_processors = manager_config.get(
+            'zone-processors', ['dynamic-zone-config']
+        )
+        self.log.info('__init__: zone_processors=%s', self.zone_processors)
+
         self.global_processors = manager_config.get('processors', [])
         self.log.info('__init__: global_processors=%s', self.global_processors)
 
@@ -259,6 +265,11 @@ class Manager(object):
                 raise ManagerException(
                     f'Incorrect processor config for {processor_name}, {processor_config.context}'
                 )
+
+        name = 'dynamic-zone-config'
+        if name not in processors:
+            processors[name] = DynamicZoneConfigProcessor(name)
+
         return processors
 
     def _config_plan_outputs(self, plan_outputs_config):
@@ -521,35 +532,19 @@ class Manager(object):
         the call and the zones returned from this function should be used
         instead.
         '''
-        for name, config in list(zones.items()):
-            if not name.startswith('*'):
-                continue
-            # we've found a dynamic config element
 
-            # find its sources
-            found_sources = sources or self._get_sources(
-                name, config, eligible_sources
-            )
-            self.log.info('sync:   dynamic zone=%s, sources=%s', name, sources)
-            for source in found_sources:
-                if not hasattr(source, 'list_zones'):
-                    raise ManagerException(
-                        f'dynamic zone={name} includes a source, {source.id}, that does not support `list_zones`'
-                    )
-                for zone_name in source.list_zones():
-                    if zone_name in zones:
-                        self.log.info(
-                            'sync:      zone=%s already in config, ignoring',
-                            zone_name,
-                        )
-                        continue
-                    self.log.info(
-                        'sync:      adding dynamic zone=%s', zone_name
-                    )
-                    zones[zone_name] = config
+        zone_processors = []
+        try:
+            for processor in self.zone_processors:
+                zone_processors.append(self.processors[processor])
+        except KeyError:
+            raise ManagerException(f'unknown zone processor: {processor}')
+
+        def get_sources(name, config):
+            return sources or self._get_sources(name, config, eligible_sources)
 
-            # remove the dynamic config element so we don't try and populate it
-            del zones[name]
+        for zone_processor in zone_processors:
+            zones = zone_processor.process_zone_config(zones, get_sources)
 
         return zones
 
diff --git a/octodns/processor/base.py b/octodns/processor/base.py
index eb584d5..52fa074 100644
--- a/octodns/processor/base.py
+++ b/octodns/processor/base.py
@@ -12,6 +12,22 @@ class BaseProcessor(object):
         # TODO: name is DEPRECATED, remove in 2.0
         self.id = self.name = name
 
+    def process_zone_config(self, zones, get_sources):
+        '''
+        Called by the Manager after loading the zone config data. Provides an
+        opportunity for the processor to modify the `Zone` configs and thus what
+        zones octoDNS will manage and the settings used for them.
+
+        - Will see `zones` after any modifications done by
+          `Processors.process_zone_config` configured to run before this one
+        - May modify `zones` directly.
+        - Will receive a callable `get_sources` that can be called as
+          `get_sources(zone_name, zone_config)` to get a list of
+          sources/providers that will be used to populate the zone.
+        - Must return `zones` which will normally be the `zones` param.
+        '''
+        return zones
+
     def process_source_zone(self, desired, sources):
         '''
         Called after all sources have completed populate. Provides an
diff --git a/octodns/processor/zone.py b/octodns/processor/zone.py
new file mode 100644
index 0000000..8574e1c
--- /dev/null
+++ b/octodns/processor/zone.py
@@ -0,0 +1,45 @@
+#
+#
+#
+
+from logging import getLogger
+
+from .base import BaseProcessor, ProcessorException
+
+
+class DynamicZoneConfigProcessor(BaseProcessor):
+    log = getLogger('DynamicZoneConfigProcessor')
+
+    def process_zone_config(self, zones, get_sources):
+        for name, config in list(zones.items()):
+            if not name.startswith('*'):
+                continue
+            # we've found a dynamic config element
+
+            # find its sources
+            found_sources = get_sources(name, config)
+
+            self.log.info(
+                'sync:   dynamic zone=%s, sources=%s', name, found_sources
+            )
+            for source in found_sources:
+                if not hasattr(source, 'list_zones'):
+                    raise ProcessorException(
+                        f'dynamic zone={name} includes a source, {source.id}, that does not support `list_zones`'
+                    )
+                for zone_name in source.list_zones():
+                    if zone_name in zones:
+                        self.log.info(
+                            'sync:      zone=%s already in config, ignoring',
+                            zone_name,
+                        )
+                        continue
+                    self.log.info(
+                        'sync:      adding dynamic zone=%s', zone_name
+                    )
+                    zones[zone_name] = config
+
+            # remove the dynamic config element so we don't try and populate it
+            del zones[name]
+
+        return zones
diff --git a/tests/config/processors-unknown-zone-processor.yaml b/tests/config/processors-unknown-zone-processor.yaml
new file mode 100644
index 0000000..fae054d
--- /dev/null
+++ b/tests/config/processors-unknown-zone-processor.yaml
@@ -0,0 +1,31 @@
+
+manager:
+  zone-processors:
+    - dynamic-zone-config
+    - this-does-not-exist
+
+providers:
+  config:
+    # This helps us get coverage when printing out provider versions
+    class: helpers.TestYamlProvider
+    directory: tests/config
+    strict_supports: False
+  dump:
+    class: octodns.provider.yaml.YamlProvider
+    directory: env/YAML_TMP_DIR
+    supports_root_ns: False
+    strict_supports: False
+
+processors:
+  # manually define the dynamic zone config, just to do it
+  dynamic-zone-config:
+    class: octodns.processor.zone.DynamicZoneConfigProcessor
+
+zones:
+  unit.tests.:
+    processors:
+    - noop
+    sources:
+    - config
+    targets:
+    - dump
diff --git a/tests/test_octodns_manager.py b/tests/test_octodns_manager.py
index 79f4318..b1bda4b 100644
--- a/tests/test_octodns_manager.py
+++ b/tests/test_octodns_manager.py
@@ -24,7 +24,7 @@ from octodns.manager import (
     ManagerException,
     _AggregateTarget,
 )
-from octodns.processor.base import BaseProcessor
+from octodns.processor.base import BaseProcessor, ProcessorException
 from octodns.record import Create, Delete, Record, Update
 from octodns.yaml import safe_load
 from octodns.zone import Zone
@@ -651,7 +651,8 @@ class TestManager(TestCase):
         # Smoke test loading a valid config
         manager = Manager(get_config_filename('processors.yaml'))
         self.assertEqual(
-            ['noop', 'test', 'global-counter'], list(manager.processors.keys())
+            ['dynamic-zone-config', 'global-counter', 'noop', 'test'],
+            sorted(manager.processors.keys()),
         )
         # make sure we got the global processor and that it's count is 0 now
         self.assertEqual(['global-counter'], manager.global_processors)
@@ -786,6 +787,22 @@ class TestManager(TestCase):
         # no plans
         self.assertFalse(plans)
 
+    def test_zone_processor(self):
+        # Smoke test loading a valid config
+        manager = Manager(
+            get_config_filename('processors-unknown-zone-processor.yaml')
+        )
+        self.assertEqual(
+            ['dynamic-zone-config', 'this-does-not-exist'],
+            manager.zone_processors,
+        )
+
+        with self.assertRaises(ManagerException) as ctx:
+            manager.sync(['unit.tests.'])
+        self.assertEqual(
+            'unknown zone processor: this-does-not-exist', str(ctx.exception)
+        )
+
     def test_try_version(self):
         manager = Manager(get_config_filename('simple.yaml'))
 
@@ -1067,7 +1084,7 @@ class TestManager(TestCase):
             get_config_filename('dynamic-config-no-list-zones.yaml')
         )
 
-        with self.assertRaises(ManagerException) as ctx:
+        with self.assertRaises(ProcessorException) as ctx:
             manager.sync()
         self.assertTrue('does not support `list_zones`' in str(ctx.exception))
 
diff --git a/tests/test_octodns_processor.py b/tests/test_octodns_processor.py
new file mode 100644
index 0000000..3959d31
--- /dev/null
+++ b/tests/test_octodns_processor.py
@@ -0,0 +1,43 @@
+#
+#
+#
+
+from unittest import TestCase
+
+from octodns.processor.base import BaseProcessor
+
+
+class BaseProcessorTest(TestCase):
+    proc = BaseProcessor('test')
+
+    def test_process_zone_config(self):
+        def get_sources(name, config):
+            return []
+
+        zones = {}
+        got = self.proc.process_zone_config(zones, get_sources)
+        self.assertIs(zones, got)
+
+    def test_process_source_zone(self):
+        desired = 42
+        got = self.proc.process_source_zone(desired, [])
+        self.assertIs(desired, got)
+
+    def test_process_target_zone(self):
+        existing = 43
+        got = self.proc.process_target_zone(existing, None)
+        self.assertIs(existing, got)
+
+    def test_process_source_and_target_zones(self):
+        desired = 42
+        existing = 43
+        got_desired, got_existing = self.proc.process_source_and_target_zones(
+            desired, existing, None
+        )
+        self.assertIs(desired, got_desired)
+        self.assertIs(existing, got_existing)
+
+    def test_process_plan(self):
+        plan = 42
+        got = self.proc.process_plan(plan, [], None)
+        self.assertIs(plan, got)
diff --git a/tests/test_octodns_processor_zone.py b/tests/test_octodns_processor_zone.py
new file mode 100644
index 0000000..9cd4b6a
--- /dev/null
+++ b/tests/test_octodns_processor_zone.py
@@ -0,0 +1,107 @@
+#
+#
+#
+
+from unittest import TestCase
+
+from octodns.processor.base import ProcessorException
+from octodns.processor.zone import DynamicZoneConfigProcessor
+
+
+class _GetSourcesMock:
+    def __init__(self, sources=[]):
+        self.data = {}
+        self.sources = sources
+        self.called = 0
+
+    def get_sources(self, name, config):
+        self.data[name] = config
+        self.called += 1
+        return self.sources
+
+
+class _ListZonesMock:
+    id = '_ListZonesMock'
+
+    def __init__(self, zones):
+        self.zones = zones
+        self.called = 0
+
+    def list_zones(self):
+        self.called += 1
+        return self.zones
+
+
+class _NoListZonesMock:
+    id = '_NoListZonesMock'
+
+
+class BaseProcessorTest(TestCase):
+    proc = DynamicZoneConfigProcessor('test')
+
+    def test_process_zone_config_empty(self):
+        mock = _GetSourcesMock()
+
+        zones = {}
+        got = self.proc.process_zone_config(zones, mock.get_sources)
+        self.assertFalse(mock.called)
+        self.assertIs(zones, got)
+
+    def test_process_zone_config_static(self):
+        mock = _GetSourcesMock()
+
+        zones = {'unit.tests.': {'key': 'value'}}
+        got = self.proc.process_zone_config(zones, mock.get_sources)
+        self.assertFalse(mock.called)
+        self.assertIs(zones, got)
+
+    def test_process_zone_config_dynamic(self):
+        lz_mock = _ListZonesMock(
+            [
+                'dynamic1.unit.tests.',
+                'dynamic2.unit.tests.',
+                'existing.unit.tests.',
+            ]
+        )
+        gs_mock = _GetSourcesMock([lz_mock])
+
+        zones = {
+            '*': {'type': 'dynamic'},
+            'unit.tests.': {'type': 'static'},
+            'existing.unit.tests.': {'type': 'exsiting'},
+        }
+        got = self.proc.process_zone_config(zones, gs_mock.get_sources)
+        self.assertEqual(1, gs_mock.called)
+        self.assertEqual({'*': {'type': 'dynamic'}}, gs_mock.data)
+        self.assertEqual(1, lz_mock.called)
+
+        self.assertIs(zones, got)
+        self.assertEqual({'type': 'dynamic'}, got['dynamic1.unit.tests.'])
+        self.assertEqual({'type': 'dynamic'}, got['dynamic2.unit.tests.'])
+
+    def test_process_zone_config_dynamic_prefix(self):
+        lz_mock = _ListZonesMock(['dyn-pre.unit.tests.'])
+        gs_mock = _GetSourcesMock([lz_mock])
+
+        zones = {'*.foo': {'type': 'dynamic-too'}}
+        got = self.proc.process_zone_config(zones, gs_mock.get_sources)
+        self.assertEqual(1, gs_mock.called)
+        self.assertEqual({'*.foo': {'type': 'dynamic-too'}}, gs_mock.data)
+        self.assertEqual(1, lz_mock.called)
+
+        self.assertIs(zones, got)
+        from pprint import pprint
+
+        pprint(got)
+        self.assertEqual({'type': 'dynamic-too'}, got['dyn-pre.unit.tests.'])
+
+    def test_process_zone_config_no_list_zones(self):
+        gs_mock = _GetSourcesMock([_NoListZonesMock()])
+
+        zones = {'*': {'type': 'dynamic'}}
+        with self.assertRaises(ProcessorException) as ctx:
+            self.proc.process_zone_config(zones, gs_mock.get_sources)
+        self.assertEqual(
+            'dynamic zone=* includes a source, _NoListZonesMock, that does not support `list_zones`',
+            str(ctx.exception),
+        )