diff --git a/CHANGELOG.md b/CHANGELOG.md
index 78685f6..a84f0ed 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,19 @@
 ## v0.9.19 - 2022-??-?? - ???
 
+#### Noteworthy changes
+
+* Added support for automatic handling of IDNA (utf-8) zones. Everything is
+  stored IDNA encoded internally. For ASCII zones that's a noop. For zones with
+  utf-8 chars they will be converted and all internals/providers will see the
+  encoded version and work with it without any knowledge of it having been
+  converted. This means that all providers will automatically support IDNA as of
+  this version. IDNA zones will generally be displayed in the logs in their
+  decoded form. Both forms should be accepted in command line arguments.
+  Providers may need to be updated to display the decoded form in their logs,
+  until then they'd display the IDNA version.
+
+#### Stuff
+
 * Addressed shortcomings with YamlProvider.SUPPORTS in that it didn't include
   dynamically registered types, was a static list that could have drifted over
   time even ignoring 3rd party types.
@@ -11,6 +25,9 @@
 * Now that it's used as it needed to be YamlProvider overrides
   Provider.supports and just always says Yes so that any dynamically registered
   types will be supported.
+* Add TtlRestrictionFilter processor for adding ttl restriction/checking
+* NameAllowlistFilter & NameRejectlistFilter implementations to support
+  filtering on record names to include/exclude records from management.
 
 ## v0.9.18 - 2022-08-14 - Subzone handling
 
diff --git a/README.md b/README.md
index 666e4b9..788619f 100644
--- a/README.md
+++ b/README.md
@@ -363,6 +363,7 @@ If you have a problem or suggestion, please [open an issue](https://github.com/o
   - [`doddo/octodns-lexicon`](https://github.com/doddo/octodns-lexicon): Use [Lexicon](https://github.com/AnalogJ/lexicon) providers as octoDNS providers.
   - [`asyncon/octoblox`](https://github.com/asyncon/octoblox): [Infoblox](https://www.infoblox.com/) provider.
   - [`sukiyaki/octodns-netbox`](https://github.com/sukiyaki/octodns-netbox): [NetBox](https://github.com/netbox-community/netbox) source.
+  - [`jcollie/octodns-netbox-dns`](https://github.com/jcollie/octodns-netbox-dns): [NetBox-DNS Plugin](https://github.com/auroraresearchlab/netbox-dns) provider. 
   - [`kompetenzbolzen/octodns-custom-provider`](https://github.com/kompetenzbolzen/octodns-custom-provider): zonefile provider & phpIPAM source.
 - **Resources.**
   - Article: [Visualising DNS records with Neo4j](https://medium.com/@costask/querying-and-visualising-octodns-records-with-neo4j-f4f72ab2d474) + code
diff --git a/octodns/cmds/report.py b/octodns/cmds/report.py
index d13e85c..03a8f11 100755
--- a/octodns/cmds/report.py
+++ b/octodns/cmds/report.py
@@ -93,7 +93,7 @@ def main():
         ]
 
     for record, futures in sorted(queries.items(), key=lambda d: d[0]):
-        stdout.write(record.fqdn)
+        stdout.write(record.decoded_fqdn)
         stdout.write(',')
         stdout.write(record._type)
         stdout.write(',')
diff --git a/octodns/idna.py b/octodns/idna.py
index bc91d46..9a079ee 100644
--- a/octodns/idna.py
+++ b/octodns/idna.py
@@ -2,35 +2,84 @@
 #
 #
 
-from idna import decode as _decode, encode as _encode
+from collections.abc import MutableMapping
+
+from idna import IDNAError as _IDNAError, decode as _decode, encode as _encode
 
 # Providers will need to to make calls to these at the appropriate points,
 # generally right before they pass names off to api calls. For an example of
 # usage see https://github.com/octodns/octodns-ns1/pull/20
 
 
+class IdnaError(Exception):
+    def __init__(self, idna_error):
+        super().__init__(str(idna_error))
+
+
 def idna_encode(name):
     # Based on https://github.com/psf/requests/pull/3695/files
     # #diff-0debbb2447ce5debf2872cb0e17b18babe3566e9d9900739e8581b355bd513f7R39
+    name = name.lower()
     try:
         name.encode('ascii')
         # No utf8 chars, just use as-is
         return name
     except UnicodeEncodeError:
-        if name.startswith('*'):
-            # idna.encode doesn't like the *
-            name = _encode(name[2:]).decode('utf-8')
-            return f'*.{name}'
-        return _encode(name).decode('utf-8')
+        try:
+            if name.startswith('*'):
+                # idna.encode doesn't like the *
+                name = _encode(name[2:]).decode('utf-8')
+                return f'*.{name}'
+            return _encode(name).decode('utf-8')
+        except _IDNAError as e:
+            raise IdnaError(e)
 
 
 def idna_decode(name):
     pieces = name.lower().split('.')
     if any(p.startswith('xn--') for p in pieces):
-        # it's idna
-        if name.startswith('*'):
-            # idna.decode doesn't like the *
-            return f'*.{_decode(name[2:])}'
-        return _decode(name)
+        try:
+            # it's idna
+            if name.startswith('*'):
+                # idna.decode doesn't like the *
+                return f'*.{_decode(name[2:])}'
+            return _decode(name)
+        except _IDNAError as e:
+            raise IdnaError(e)
     # not idna, just return as-is
     return name
+
+
+class IdnaDict(MutableMapping):
+    '''A dict type that is insensitive to case and utf-8/idna encoded strings'''
+
+    def __init__(self, data=None):
+        self._data = dict()
+        if data is not None:
+            self.update(data)
+
+    def __setitem__(self, k, v):
+        self._data[idna_encode(k)] = v
+
+    def __getitem__(self, k):
+        return self._data[idna_encode(k)]
+
+    def __delitem__(self, k):
+        del self._data[idna_encode(k)]
+
+    def __iter__(self):
+        return iter(self._data)
+
+    def __len__(self):
+        return len(self._data)
+
+    def decoded_keys(self):
+        for key in self.keys():
+            yield idna_decode(key)
+
+    def decoded_items(self):
+        for key, value in self.items():
+            yield (idna_decode(key), value)
+
+    def __repr__(self):
+        return self._data.__repr__()
diff --git a/octodns/manager.py b/octodns/manager.py
index 19c5122..c58074a 100644
--- a/octodns/manager.py
+++ b/octodns/manager.py
@@ -17,6 +17,7 @@ from sys import stdout
 import logging
 
 from . import __VERSION__
+from .idna import IdnaDict, idna_decode, idna_encode
 from .provider.base import BaseProvider
 from .provider.plan import Plan
 from .provider.yaml import SplitYamlProvider, YamlProvider
@@ -111,30 +112,76 @@ class Manager(object):
             '__init__: config_file=%s (octoDNS %s)', config_file, version
         )
 
+        self._configured_sub_zones = None
+
         # Read our config file
         with open(config_file, 'r') as fh:
             self.config = safe_load(fh, enforce_order=False)
 
+        zones = self.config['zones']
+        self.config['zones'] = self._config_zones(zones)
+
         manager_config = self.config.get('manager', {})
+        self._executor = self._config_executor(manager_config, max_workers)
+        self.include_meta = self._config_include_meta(
+            manager_config, include_meta
+        )
+
+        providers_config = self.config['providers']
+        self.providers = self._config_providers(providers_config)
+
+        processors_config = self.config.get('processors', {})
+        self.processors = self._config_processors(processors_config)
+
+        plan_outputs_config = manager_config.get(
+            'plan_outputs',
+            {
+                '_logger': {
+                    'class': 'octodns.provider.plan.PlanLogger',
+                    'level': 'info',
+                }
+            },
+        )
+        self.plan_outputs = self._config_plan_outputs(plan_outputs_config)
+
+    def _config_zones(self, zones):
+        # record the set of configured zones we have as they are
+        configured_zones = set([z.lower() for z in zones.keys()])
+        # walk the configured zones
+        for name in configured_zones:
+            if 'xn--' not in name:
+                continue
+            # this is an IDNA format zone name
+            decoded = idna_decode(name)
+            # do we also have a config for its utf-8
+            if decoded in configured_zones:
+                raise ManagerException(
+                    f'"{decoded}" configured both in utf-8 and idna "{name}"'
+                )
+
+        # convert the zones portion of things into an IdnaDict
+        return IdnaDict(zones)
+
+    def _config_executor(self, manager_config, max_workers=None):
         max_workers = (
             manager_config.get('max_workers', 1)
             if max_workers is None
             else max_workers
         )
-        self.log.info('__init__:   max_workers=%d', max_workers)
+        self.log.info('_config_executor: max_workers=%d', max_workers)
         if max_workers > 1:
-            self._executor = ThreadPoolExecutor(max_workers=max_workers)
-        else:
-            self._executor = MainThreadExecutor()
+            return ThreadPoolExecutor(max_workers=max_workers)
+        return MainThreadExecutor()
 
-        self.include_meta = include_meta or manager_config.get(
-            'include_meta', False
-        )
-        self.log.info('__init__:   include_meta=%s', self.include_meta)
+    def _config_include_meta(self, manager_config, include_meta=False):
+        include_meta = include_meta or manager_config.get('include_meta', False)
+        self.log.info('_config_include_meta: include_meta=%s', include_meta)
+        return include_meta
 
-        self.log.debug('__init__:   configuring providers')
-        self.providers = {}
-        for provider_name, provider_config in self.config['providers'].items():
+    def _config_providers(self, providers_config):
+        self.log.debug('_config_providers: configuring providers')
+        providers = {}
+        for provider_name, provider_config in providers_config.items():
             # Get our class and remove it from the provider_config
             try:
                 _class = provider_config.pop('class')
@@ -146,7 +193,7 @@ class Manager(object):
             _class, module, version = self._get_named_class('provider', _class)
             kwargs = self._build_kwargs(provider_config)
             try:
-                self.providers[provider_name] = _class(provider_name, **kwargs)
+                providers[provider_name] = _class(provider_name, **kwargs)
                 self.log.info(
                     '__init__: provider=%s (%s %s)',
                     provider_name,
@@ -159,10 +206,11 @@ class Manager(object):
                     'Incorrect provider config for ' + provider_name
                 )
 
-        self.processors = {}
-        for processor_name, processor_config in self.config.get(
-            'processors', {}
-        ).items():
+        return providers
+
+    def _config_processors(self, processors_config):
+        processors = {}
+        for processor_name, processor_config in processors_config.items():
             try:
                 _class = processor_config.pop('class')
             except KeyError:
@@ -173,9 +221,7 @@ class Manager(object):
             _class, module, version = self._get_named_class('processor', _class)
             kwargs = self._build_kwargs(processor_config)
             try:
-                self.processors[processor_name] = _class(
-                    processor_name, **kwargs
-                )
+                processors[processor_name] = _class(processor_name, **kwargs)
                 self.log.info(
                     '__init__: processor=%s (%s %s)',
                     processor_name,
@@ -187,18 +233,11 @@ class Manager(object):
                 raise ManagerException(
                     'Incorrect processor config for ' + processor_name
                 )
+        return processors
 
-        self.plan_outputs = {}
-        plan_outputs = manager_config.get(
-            'plan_outputs',
-            {
-                '_logger': {
-                    'class': 'octodns.provider.plan.PlanLogger',
-                    'level': 'info',
-                }
-            },
-        )
-        for plan_output_name, plan_output_config in plan_outputs.items():
+    def _config_plan_outputs(self, plan_outputs_config):
+        plan_outputs = {}
+        for plan_output_name, plan_output_config in plan_outputs_config.items():
             try:
                 _class = plan_output_config.pop('class')
             except KeyError:
@@ -211,7 +250,7 @@ class Manager(object):
             )
             kwargs = self._build_kwargs(plan_output_config)
             try:
-                self.plan_outputs[plan_output_name] = _class(
+                plan_outputs[plan_output_name] = _class(
                     plan_output_name, **kwargs
                 )
                 # Don't print out version info for the default output
@@ -227,8 +266,7 @@ class Manager(object):
                 raise ManagerException(
                     'Incorrect plan_output config for ' + plan_output_name
                 )
-
-        self._configured_sub_zones = None
+        return plan_outputs
 
     def _try_version(self, module_name, module=None, version=None):
         try:
@@ -300,10 +338,21 @@ class Manager(object):
         return kwargs
 
     def configured_sub_zones(self, zone_name):
+        '''
+        Accepts either UTF-8 or IDNA encoded zone name and returns the list of
+        any configured sub-zones in IDNA form. E.g. for the following
+        configured zones:
+          some.com.
+          other.some.com.
+          deep.thing.some.com.
+        It would return
+          other
+          deep.thing
+        '''
         if self._configured_sub_zones is None:
             # First time through we compute all the sub-zones
 
-            configured_sub_zones = {}
+            configured_sub_zones = IdnaDict()
 
             # Get a list of all of our zone names. Sort them from shortest to
             # longest so that parents will always come before their subzones
@@ -341,10 +390,12 @@ class Manager(object):
         lenient=False,
     ):
 
-        self.log.debug(
-            'sync:   populating, zone=%s, lenient=%s', zone_name, lenient
-        )
         zone = Zone(zone_name, sub_zones=self.configured_sub_zones(zone_name))
+        self.log.debug(
+            'sync:   populating, zone=%s, lenient=%s',
+            zone.decoded_name,
+            lenient,
+        )
 
         if desired:
             # This is an alias zone, rather than populate it we'll copy the
@@ -368,7 +419,7 @@ class Manager(object):
         for processor in processors:
             zone = processor.process_source_zone(zone, sources=sources)
 
-        self.log.debug('sync:   planning, zone=%s', zone_name)
+        self.log.debug('sync:   planning, zone=%s', zone.decoded_name)
         plans = []
 
         for target in targets:
@@ -424,40 +475,29 @@ class Manager(object):
             getattr(plan_output_fh, 'name', plan_output_fh.__class__.__name__),
         )
 
-        zones = self.config['zones'].items()
+        zones = self.config['zones']
         if eligible_zones:
-            zones = [z for z in zones if z[0] in eligible_zones]
+            zones = IdnaDict({n: zones.get(n) for n in eligible_zones})
 
         aliased_zones = {}
         futures = []
-        for zone_name, config in zones:
-            self.log.info('sync:   zone=%s', zone_name)
+        for zone_name, config in zones.items():
+            decoded_zone_name = idna_decode(zone_name)
+            self.log.info('sync:   zone=%s', decoded_zone_name)
             if 'alias' in config:
                 source_zone = config['alias']
 
                 # Check that the source zone is defined.
                 if source_zone not in self.config['zones']:
-                    self.log.error(
-                        f'Invalid alias zone {zone_name}, '
-                        f'target {source_zone} does not exist'
-                    )
-                    raise ManagerException(
-                        f'Invalid alias zone {zone_name}: '
-                        f'source zone {source_zone} does '
-                        'not exist'
-                    )
+                    msg = f'Invalid alias zone {decoded_zone_name}: source zone {idna_decode(source_zone)} does not exist'
+                    self.log.error(msg)
+                    raise ManagerException(msg)
 
                 # Check that the source zone is not an alias zone itself.
                 if 'alias' in self.config['zones'][source_zone]:
-                    self.log.error(
-                        f'Invalid alias zone {zone_name}, '
-                        f'target {source_zone} is an alias zone'
-                    )
-                    raise ManagerException(
-                        f'Invalid alias zone {zone_name}: '
-                        f'source zone {source_zone} is an '
-                        'alias zone'
-                    )
+                    msg = f'Invalid alias zone {decoded_zone_name}: source zone {idna_decode(source_zone)} is an alias zone'
+                    self.log.error(msg)
+                    raise ManagerException(msg)
 
                 aliased_zones[zone_name] = source_zone
                 continue
@@ -466,12 +506,16 @@ class Manager(object):
             try:
                 sources = config['sources']
             except KeyError:
-                raise ManagerException(f'Zone {zone_name} is missing sources')
+                raise ManagerException(
+                    f'Zone {decoded_zone_name} is missing sources'
+                )
 
             try:
                 targets = config['targets']
             except KeyError:
-                raise ManagerException(f'Zone {zone_name} is missing targets')
+                raise ManagerException(
+                    f'Zone {decoded_zone_name} is missing targets'
+                )
 
             processors = config.get('processors', [])
 
@@ -500,7 +544,8 @@ class Manager(object):
                 processors = collected
             except KeyError:
                 raise ManagerException(
-                    f'Zone {zone_name}, unknown ' f'processor: {processor}'
+                    f'Zone {decoded_zone_name}, unknown '
+                    f'processor: {processor}'
                 )
 
             try:
@@ -513,7 +558,7 @@ class Manager(object):
                 sources = collected
             except KeyError:
                 raise ManagerException(
-                    f'Zone {zone_name}, unknown ' f'source: {source}'
+                    f'Zone {decoded_zone_name}, unknown ' f'source: {source}'
                 )
 
             try:
@@ -528,7 +573,7 @@ class Manager(object):
                 targets = trgs
             except KeyError:
                 raise ManagerException(
-                    f'Zone {zone_name}, unknown ' f'target: {target}'
+                    f'Zone {decoded_zone_name}, unknown ' f'target: {target}'
                 )
 
             futures.append(
@@ -560,7 +605,7 @@ class Manager(object):
                 desired_config = desired[zone_source]
             except KeyError:
                 raise ManagerException(
-                    f'Zone {zone_name} cannot be sync '
+                    f'Zone {idna_decode(zone_name)} cannot be synced '
                     f'without zone {zone_source} sinced '
                     'it is aliased'
                 )
@@ -602,7 +647,7 @@ class Manager(object):
         self.log.debug('sync:   applying')
         zones = self.config['zones']
         for target, plan in plans:
-            zone_name = plan.existing.name
+            zone_name = plan.existing.decoded_name
             if zones[zone_name].get('always-dry-run', False):
                 self.log.info(
                     'sync: zone=%s skipping always-dry-run', zone_name
@@ -718,7 +763,9 @@ class Manager(object):
         target.apply(plan)
 
     def validate_configs(self):
+        # TODO: this code can probably be shared with stuff in sync
         for zone_name, config in self.config['zones'].items():
+            decoded_zone_name = idna_decode(zone_name)
             zone = Zone(zone_name, self.configured_sub_zones(zone_name))
 
             source_zone = config.get('alias')
@@ -726,7 +773,7 @@ class Manager(object):
                 if source_zone not in self.config['zones']:
                     self.log.exception('Invalid alias zone')
                     raise ManagerException(
-                        f'Invalid alias zone {zone_name}: '
+                        f'Invalid alias zone {decoded_zone_name}: '
                         f'source zone {source_zone} does '
                         'not exist'
                     )
@@ -734,7 +781,7 @@ class Manager(object):
                 if 'alias' in self.config['zones'][source_zone]:
                     self.log.exception('Invalid alias zone')
                     raise ManagerException(
-                        f'Invalid alias zone {zone_name}: '
+                        f'Invalid alias zone {decoded_zone_name}: '
                         'source zone {source_zone} is an '
                         'alias zone'
                     )
@@ -748,7 +795,9 @@ class Manager(object):
             try:
                 sources = config['sources']
             except KeyError:
-                raise ManagerException(f'Zone {zone_name} is missing sources')
+                raise ManagerException(
+                    f'Zone {decoded_zone_name} is missing sources'
+                )
 
             try:
                 # rather than using a list comprehension, we break this
@@ -760,7 +809,7 @@ class Manager(object):
                 sources = collected
             except KeyError:
                 raise ManagerException(
-                    f'Zone {zone_name}, unknown source: ' + source
+                    f'Zone {decoded_zone_name}, unknown source: ' + source
                 )
 
             for source in sources:
@@ -775,17 +824,20 @@ class Manager(object):
                     collected.append(self.processors[processor])
             except KeyError:
                 raise ManagerException(
-                    f'Zone {zone_name}, unknown ' f'processor: {processor}'
+                    f'Zone {decoded_zone_name}, unknown '
+                    f'processor: {processor}'
                 )
 
     def get_zone(self, zone_name):
         if not zone_name[-1] == '.':
             raise ManagerException(
-                f'Invalid zone name {zone_name}, missing ending dot'
+                f'Invalid zone name {idna_decode(zone_name)}, missing ending dot'
             )
 
-        for name, config in self.config['zones'].items():
-            if name == zone_name:
-                return Zone(name, self.configured_sub_zones(name))
+        zone = self.config['zones'].get(zone_name)
+        if zone:
+            return Zone(
+                idna_encode(zone_name), self.configured_sub_zones(zone_name)
+            )
 
-        raise ManagerException(f'Unknown zone name {zone_name}')
+        raise ManagerException(f'Unknown zone name {idna_decode(zone_name)}')
diff --git a/octodns/processor/base.py b/octodns/processor/base.py
index ac5c155..c6c368e 100644
--- a/octodns/processor/base.py
+++ b/octodns/processor/base.py
@@ -10,6 +10,10 @@ from __future__ import (
 )
 
 
+class ProcessorException(Exception):
+    pass
+
+
 class BaseProcessor(object):
     def __init__(self, name):
         self.name = name
diff --git a/octodns/processor/filter.py b/octodns/processor/filter.py
index f3aabf5..fdbd8ec 100644
--- a/octodns/processor/filter.py
+++ b/octodns/processor/filter.py
@@ -9,6 +9,8 @@ from __future__ import (
     unicode_literals,
 )
 
+from re import compile as re_compile
+
 from .base import BaseProcessor
 
 
@@ -19,8 +21,8 @@ class TypeAllowlistFilter(BaseProcessor):
 
     processors:
       only-a-and-aaaa:
-        class: octodns.processor.filter.TypeRejectlistFilter
-        rejectlist:
+        class: octodns.processor.filter.TypeAllowlistFilter
+        allowlist:
           - A
           - AAAA
 
@@ -35,7 +37,7 @@ class TypeAllowlistFilter(BaseProcessor):
     '''
 
     def __init__(self, name, allowlist):
-        super(TypeAllowlistFilter, self).__init__(name)
+        super().__init__(name)
         self.allowlist = set(allowlist)
 
     def _process(self, zone, *args, **kwargs):
@@ -71,7 +73,7 @@ class TypeRejectlistFilter(BaseProcessor):
     '''
 
     def __init__(self, name, rejectlist):
-        super(TypeRejectlistFilter, self).__init__(name)
+        super().__init__(name)
         self.rejectlist = set(rejectlist)
 
     def _process(self, zone, *args, **kwargs):
@@ -83,3 +85,113 @@ class TypeRejectlistFilter(BaseProcessor):
 
     process_source_zone = _process
     process_target_zone = _process
+
+
+class _NameBaseFilter(BaseProcessor):
+    def __init__(self, name, _list):
+        super().__init__(name)
+        exact = set()
+        regex = []
+        for pattern in _list:
+            if pattern.startswith('/'):
+                regex.append(re_compile(pattern[1:-1]))
+            else:
+                exact.add(pattern)
+        self.exact = exact
+        self.regex = regex
+
+
+class NameAllowlistFilter(_NameBaseFilter):
+    '''Only manage records with names that match the provider patterns
+
+    Example usage:
+
+    processors:
+      only-these:
+        class: octodns.processor.filter.NameAllowlistFilter
+        allowlist:
+          # exact string match
+          - www
+          # contains/substring match
+          - /substring/
+          # regex pattern match
+          - /some-pattern-\\d\\+/
+          # regex - anchored so has to match start to end
+          - /^start-.+-end$/
+
+    zones:
+      exxampled.com.:
+        sources:
+          - config
+        processors:
+          - only-these
+        targets:
+          - route53
+    '''
+
+    def __init__(self, name, allowlist):
+        super().__init__(name, allowlist)
+
+    def _process(self, zone, *args, **kwargs):
+        for record in zone.records:
+            name = record.name
+            if name in self.exact:
+                continue
+            elif any(r.search(name) for r in self.regex):
+                continue
+
+            zone.remove_record(record)
+
+        return zone
+
+    process_source_zone = _process
+    process_target_zone = _process
+
+
+class NameRejectlistFilter(_NameBaseFilter):
+    '''Reject managing records with names that match the provider patterns
+
+    Example usage:
+
+    processors:
+      not-these:
+        class: octodns.processor.filter.NameRejectlistFilter
+        rejectlist:
+          # exact string match
+          - www
+          # contains/substring match
+          - /substring/
+          # regex pattern match
+          - /some-pattern-\\d\\+/
+          # regex - anchored so has to match start to end
+          - /^start-.+-end$/
+
+    zones:
+      exxampled.com.:
+        sources:
+          - config
+        processors:
+          - not-these
+        targets:
+          - route53
+    '''
+
+    def __init__(self, name, rejectlist):
+        super().__init__(name, rejectlist)
+
+    def _process(self, zone, *args, **kwargs):
+        for record in zone.records:
+            name = record.name
+            if name in self.exact:
+                zone.remove_record(record)
+                continue
+
+            for regex in self.regex:
+                if regex.search(name):
+                    zone.remove_record(record)
+                    break
+
+        return zone
+
+    process_source_zone = _process
+    process_target_zone = _process
diff --git a/octodns/processor/restrict.py b/octodns/processor/restrict.py
new file mode 100644
index 0000000..1903995
--- /dev/null
+++ b/octodns/processor/restrict.py
@@ -0,0 +1,83 @@
+#
+#
+#
+
+from __future__ import (
+    absolute_import,
+    division,
+    print_function,
+    unicode_literals,
+)
+
+from .base import BaseProcessor, ProcessorException
+
+
+class RestrictionException(ProcessorException):
+    pass
+
+
+class TtlRestrictionFilter(BaseProcessor):
+    '''
+    Ensure that configured TTLs are between a configured minimum and maximum or
+    in an allowed set of values.
+
+    The default minimum is 1 (the behavior of 0 is undefined spec-wise) and the
+    default maximum is 604800 (seven days.) allowed_ttls is only used when
+    explicitly configured and min and max are ignored in that case.
+
+    Example usage:
+
+    processors:
+      min-max-ttl:
+        class: octodns.processor.restrict.TtlRestrictionFilter
+        min_ttl: 60
+        max_ttl: 3600
+        # allowed_ttls: [300, 900, 3600]
+
+    zones:
+      exxampled.com.:
+        sources:
+          - config
+        processors:
+          - min-max-ttl
+        targets:
+          - azure
+
+    The restriction can be skipped for specific records by setting the lenient
+    flag, e.g.
+
+    a:
+      octodns:
+        lenient: true
+      ttl: 0
+      value: 1.2.3.4
+
+    The higher level lenient flags are not checked as it would make more sense
+    to just avoid enabling the processor in those cases.
+    '''
+
+    SEVEN_DAYS = 60 * 60 * 24 * 7
+
+    def __init__(self, name, min_ttl=1, max_ttl=SEVEN_DAYS, allowed_ttls=None):
+        super().__init__(name)
+        self.min_ttl = min_ttl
+        self.max_ttl = max_ttl
+        self.allowed_ttls = set(allowed_ttls) if allowed_ttls else None
+
+    def process_source_zone(self, zone, *args, **kwargs):
+        for record in zone.records:
+            if record._octodns.get('lenient'):
+                continue
+            if self.allowed_ttls and record.ttl not in self.allowed_ttls:
+                raise RestrictionException(
+                    f'{record.fqdn} ttl={record.ttl} not an allowed value, allowed_ttls={self.allowed_ttls}'
+                )
+            elif record.ttl < self.min_ttl:
+                raise RestrictionException(
+                    f'{record.fqdn} ttl={record.ttl} too low, min_ttl={self.min_ttl}'
+                )
+            elif record.ttl > self.max_ttl:
+                raise RestrictionException(
+                    f'{record.fqdn} ttl={record.ttl} too high, max_ttl={self.max_ttl}'
+                )
+        return zone
diff --git a/octodns/provider/base.py b/octodns/provider/base.py
index 88f37c9..0ef52b4 100644
--- a/octodns/provider/base.py
+++ b/octodns/provider/base.py
@@ -114,7 +114,7 @@ class BaseProvider(BaseSource):
                 self.log.warning(
                     'root NS record supported, but no record '
                     'is configured for %s',
-                    desired.name,
+                    desired.decoded_name,
                 )
         else:
             if record:
@@ -179,7 +179,7 @@ class BaseProvider(BaseSource):
         self.log.warning('%s; %s', msg, fallback)
 
     def plan(self, desired, processors=[]):
-        self.log.info('plan: desired=%s', desired.name)
+        self.log.info('plan: desired=%s', desired.decoded_name)
 
         existing = Zone(desired.name, desired.sub_zones)
         exists = self.populate(existing, target=True, lenient=True)
@@ -246,7 +246,7 @@ class BaseProvider(BaseSource):
             self.log.info('apply: disabled')
             return 0
 
-        zone_name = plan.desired.name
+        zone_name = plan.desired.decoded_name
         num_changes = len(plan.changes)
         self.log.info('apply: making %d changes to %s', num_changes, zone_name)
         self._apply(plan)
diff --git a/octodns/provider/plan.py b/octodns/provider/plan.py
index b562579..2e2fc17 100644
--- a/octodns/provider/plan.py
+++ b/octodns/provider/plan.py
@@ -165,8 +165,8 @@ class PlanLogger(_PlanOutput):
         if plans:
             current_zone = None
             for target, plan in plans:
-                if plan.desired.name != current_zone:
-                    current_zone = plan.desired.name
+                if plan.desired.decoded_name != current_zone:
+                    current_zone = plan.desired.decoded_name
                     buf.write(hr)
                     buf.write('* ')
                     buf.write(current_zone)
@@ -215,8 +215,8 @@ class PlanMarkdown(_PlanOutput):
         if plans:
             current_zone = None
             for target, plan in plans:
-                if plan.desired.name != current_zone:
-                    current_zone = plan.desired.name
+                if plan.desired.decoded_name != current_zone:
+                    current_zone = plan.desired.decoded_name
                     fh.write('## ')
                     fh.write(current_zone)
                     fh.write('\n\n')
@@ -276,8 +276,8 @@ class PlanHtml(_PlanOutput):
         if plans:
             current_zone = None
             for target, plan in plans:
-                if plan.desired.name != current_zone:
-                    current_zone = plan.desired.name
+                if plan.desired.decoded_name != current_zone:
+                    current_zone = plan.desired.decoded_name
                     fh.write('<h2>')
                     fh.write(current_zone)
                     fh.write('</h2>\n')
diff --git a/octodns/provider/yaml.py b/octodns/provider/yaml.py
index 5044332..da148db 100644
--- a/octodns/provider/yaml.py
+++ b/octodns/provider/yaml.py
@@ -17,6 +17,7 @@ import logging
 from ..record import Record
 from ..yaml import safe_load, safe_dump
 from .base import BaseProvider
+from . import ProviderException
 
 
 class YamlProvider(BaseProvider):
@@ -192,7 +193,7 @@ class YamlProvider(BaseProvider):
     def populate(self, zone, target=False, lenient=False):
         self.log.debug(
             'populate: name=%s, target=%s, lenient=%s',
-            zone.name,
+            zone.decoded_name,
             target,
             lenient,
         )
@@ -203,7 +204,23 @@ class YamlProvider(BaseProvider):
             return False
 
         before = len(zone.records)
-        filename = join(self.directory, f'{zone.name}yaml')
+        utf8_filename = join(self.directory, f'{zone.decoded_name}yaml')
+        idna_filename = join(self.directory, f'{zone.name}yaml')
+
+        # we prefer utf8
+        if isfile(utf8_filename):
+            if utf8_filename != idna_filename and isfile(idna_filename):
+                raise ProviderException(
+                    f'Both UTF-8 "{utf8_filename}" and IDNA "{idna_filename}" exist for {zone.decoded_name}'
+                )
+            filename = utf8_filename
+        else:
+            self.log.warning(
+                'populate: "%s" does not exist, falling back to try idna version "%s"',
+                utf8_filename,
+                idna_filename,
+            )
+            filename = idna_filename
         self._populate_from_file(filename, zone, lenient)
 
         self.log.info(
@@ -216,7 +233,9 @@ class YamlProvider(BaseProvider):
         desired = plan.desired
         changes = plan.changes
         self.log.debug(
-            '_apply: zone=%s, len(changes)=%d', desired.name, len(changes)
+            '_apply: zone=%s, len(changes)=%d',
+            desired.decoded_name,
+            len(changes),
         )
         # Since we don't have existing we'll only see creates
         records = [c.new for c in changes]
@@ -231,7 +250,8 @@ class YamlProvider(BaseProvider):
                 del d['ttl']
             if record._octodns:
                 d['octodns'] = record._octodns
-            data[record.name].append(d)
+            # we want to output the utf-8 version of the name
+            data[record.decoded_name].append(d)
 
         # Flatten single element lists
         for k in data.keys():
@@ -244,10 +264,10 @@ class YamlProvider(BaseProvider):
         self._do_apply(desired, data)
 
     def _do_apply(self, desired, data):
-        filename = join(self.directory, f'{desired.name}yaml')
+        filename = join(self.directory, f'{desired.decoded_name}yaml')
         self.log.debug('_apply:   writing filename=%s', filename)
         with open(filename, 'w') as fh:
-            safe_dump(dict(data), fh)
+            safe_dump(dict(data), fh, allow_unicode=True)
 
 
 def _list_all_yaml_files(directory):
diff --git a/octodns/record/__init__.py b/octodns/record/__init__.py
index 287ebe2..a0c83bb 100644
--- a/octodns/record/__init__.py
+++ b/octodns/record/__init__.py
@@ -16,6 +16,7 @@ import re
 from fqdn import FQDN
 
 from ..equality import EqualityTupleMixin
+from ..idna import IdnaError, idna_decode, idna_encode
 from .geo import GeoCodes
 
 
@@ -77,7 +78,7 @@ class ValidationError(RecordException):
     @classmethod
     def build_message(cls, fqdn, reasons):
         reasons = '\n  - '.join(reasons)
-        return f'Invalid record {fqdn}\n  - {reasons}'
+        return f'Invalid record {idna_decode(fqdn)}\n  - {reasons}'
 
     def __init__(self, fqdn, reasons):
         super(Exception, self).__init__(self.build_message(fqdn, reasons))
@@ -108,17 +109,23 @@ class Record(EqualityTupleMixin):
 
     @classmethod
     def new(cls, zone, name, data, source=None, lenient=False):
-        name = str(name).lower()
+        reasons = []
+        try:
+            name = idna_encode(str(name))
+        except IdnaError as e:
+            # convert the error into a reason
+            reasons.append(str(e))
+            name = str(name)
         fqdn = f'{name}.{zone.name}' if name else zone.name
         try:
             _type = data['type']
         except KeyError:
-            raise Exception(f'Invalid record {fqdn}, missing type')
+            raise Exception(f'Invalid record {idna_decode(fqdn)}, missing type')
         try:
             _class = cls._CLASSES[_type]
         except KeyError:
             raise Exception(f'Unknown record type: "{_type}"')
-        reasons = _class.validate(name, fqdn, data)
+        reasons.extend(_class.validate(name, fqdn, data))
         try:
             lenient |= data['octodns']['lenient']
         except KeyError:
@@ -138,7 +145,7 @@ class Record(EqualityTupleMixin):
         n = len(fqdn)
         if n > 253:
             reasons.append(
-                f'invalid fqdn, "{fqdn}" is too long at {n} '
+                f'invalid fqdn, "{idna_decode(fqdn)}" is too long at {n} '
                 'chars, max is 253'
             )
         for label in name.split('.'):
@@ -148,6 +155,7 @@ class Record(EqualityTupleMixin):
                     f'invalid label, "{label}" is too long at {n}'
                     ' chars, max is 63'
                 )
+        # TODO: look at the idna lib for a lot more potential validations...
         try:
             ttl = int(data['ttl'])
             if ttl < 0:
@@ -166,15 +174,20 @@ class Record(EqualityTupleMixin):
         return reasons
 
     def __init__(self, zone, name, data, source=None):
+        self.zone = zone
+        if name:
+            # internally everything is idna
+            self.name = idna_encode(str(name))
+            # we'll keep a decoded version around for logs and errors
+            self.decoded_name = idna_decode(self.name)
+        else:
+            self.name = self.decoded_name = name
         self.log.debug(
             '__init__: zone.name=%s, type=%11s, name=%s',
-            zone.name,
+            zone.decoded_name,
             self.__class__.__name__,
-            name,
+            self.decoded_name,
         )
-        self.zone = zone
-        # force everything lower-case just to be safe
-        self.name = str(name).lower() if name else name
         self.source = source
         self.ttl = int(data['ttl'])
 
@@ -189,10 +202,18 @@ class Record(EqualityTupleMixin):
 
     @property
     def fqdn(self):
+        # TODO: these should be calculated and set in __init__ rather than on
+        # each use
         if self.name:
             return f'{self.name}.{self.zone.name}'
         return self.zone.name
 
+    @property
+    def decoded_fqdn(self):
+        if self.decoded_name:
+            return f'{self.decoded_name}.{self.zone.decoded_name}'
+        return self.zone.decoded_name
+
     @property
     def ignored(self):
         return self._octodns.get('ignored', False)
@@ -354,7 +375,7 @@ class ValuesMixin(object):
     def __repr__(self):
         values = "', '".join([str(v) for v in self.values])
         klass = self.__class__.__name__
-        return f"<{klass} {self._type} {self.ttl}, {self.fqdn}, ['{values}']>"
+        return f"<{klass} {self._type} {self.ttl}, {self.decoded_fqdn}, ['{values}']>"
 
 
 class _GeoMixin(ValuesMixin):
@@ -404,7 +425,7 @@ class _GeoMixin(ValuesMixin):
         if self.geo:
             klass = self.__class__.__name__
             return (
-                f'<{klass} {self._type} {self.ttl}, {self.fqdn}, '
+                f'<{klass} {self._type} {self.ttl}, {self.decoded_fqdn}, '
                 f'{self.values}, {self.geo}>'
             )
         return super(_GeoMixin, self).__repr__()
@@ -436,7 +457,7 @@ class ValueMixin(object):
 
     def __repr__(self):
         klass = self.__class__.__name__
-        return f'<{klass} {self._type} {self.ttl}, {self.fqdn}, {self.value}>'
+        return f'<{klass} {self._type} {self.ttl}, {self.decoded_fqdn}, {self.value}>'
 
 
 class _DynamicPool(object):
@@ -764,7 +785,7 @@ class _DynamicMixin(object):
 
             klass = self.__class__.__name__
             return (
-                f'<{klass} {self._type} {self.ttl}, {self.fqdn}, '
+                f'<{klass} {self._type} {self.ttl}, {self.decoded_fqdn}, '
                 f'{values}, {self.dynamic}>'
             )
         return super(_DynamicMixin, self).__repr__()
diff --git a/octodns/zone.py b/octodns/zone.py
index 4c818e8..32daa4b 100644
--- a/octodns/zone.py
+++ b/octodns/zone.py
@@ -13,6 +13,7 @@ from collections import defaultdict
 from logging import getLogger
 import re
 
+from .idna import idna_decode, idna_encode
 from .record import Create, Delete
 
 
@@ -34,11 +35,15 @@ class Zone(object):
     def __init__(self, name, sub_zones):
         if not name[-1] == '.':
             raise Exception(f'Invalid zone name {name}, missing ending dot')
-        # Force everything to lowercase just to be safe
-        self.name = str(name).lower() if name else name
+        # internally everything is idna
+        self.name = idna_encode(str(name)) if name else name
+        # we'll keep a decoded version around for logs and errors
+        self.decoded_name = idna_decode(self.name)
         self.sub_zones = sub_zones
         # We're grouping by node, it allows us to efficiently search for
-        # duplicates and detect when CNAMEs co-exist with other records
+        # duplicates and detect when CNAMEs co-exist with other records. Also
+        # node that we always store things with Record.name which will be idna
+        # encoded thus we don't have to deal with idna/utf8 collisions
         self._records = defaultdict(set)
         self._root_ns = None
         # optional leading . to match empty hostname
@@ -283,4 +288,4 @@ class Zone(object):
         return copy
 
     def __repr__(self):
-        return f'Zone<{self.name}>'
+        return f'Zone<{self.decoded_name}>'
diff --git a/tests/test_octodns_idna.py b/tests/test_octodns_idna.py
index 0c6b125..7e1eb68 100644
--- a/tests/test_octodns_idna.py
+++ b/tests/test_octodns_idna.py
@@ -11,7 +11,7 @@ from __future__ import (
 
 from unittest import TestCase
 
-from octodns.idna import idna_decode, idna_encode
+from octodns.idna import IdnaDict, IdnaError, idna_decode, idna_encode
 
 
 class TestIdna(TestCase):
@@ -56,5 +56,107 @@ class TestIdna(TestCase):
         self.assertIdna('bleep_bloop.foo_bar.pl.', 'bleep_bloop.foo_bar.pl.')
 
     def test_case_insensitivity(self):
-        # Shouldn't be hit by octoDNS use cases, but checked anyway
         self.assertEqual('zajęzyk.pl.', idna_decode('XN--ZAJZYK-Y4A.PL.'))
+        self.assertEqual('xn--zajzyk-y4a.pl.', idna_encode('ZajęzyK.Pl.'))
+
+    def test_repeated_encode_decoded(self):
+        self.assertEqual(
+            'zajęzyk.pl.', idna_decode(idna_decode('xn--zajzyk-y4a.pl.'))
+        )
+        self.assertEqual(
+            'xn--zajzyk-y4a.pl.', idna_encode(idna_encode('zajęzyk.pl.'))
+        )
+
+    def test_exception_translation(self):
+        with self.assertRaises(IdnaError) as ctx:
+            idna_encode('déjà..vu.')
+        self.assertEqual('Empty Label', str(ctx.exception))
+
+        with self.assertRaises(IdnaError) as ctx:
+            idna_decode('xn--djvu-1na6c..com.')
+        self.assertEqual('Empty Label', str(ctx.exception))
+
+
+class TestIdnaDict(TestCase):
+    plain = 'testing.tests.'
+    almost = 'tésting.tests.'
+    utf8 = 'déjà.vu.'
+
+    normal = {plain: 42, almost: 43, utf8: 44}
+
+    def test_basics(self):
+        d = IdnaDict()
+
+        # plain ascii
+        d[self.plain] = 42
+        self.assertEqual(42, d[self.plain])
+
+        # almost the same, single utf-8 char
+        d[self.almost] = 43
+        # fetch as utf-8
+        self.assertEqual(43, d[self.almost])
+        # fetch as idna
+        self.assertEqual(43, d[idna_encode(self.almost)])
+        # plain is stil there, unchanged
+        self.assertEqual(42, d[self.plain])
+
+        # lots of utf8
+        d[self.utf8] = 44
+        self.assertEqual(44, d[self.utf8])
+        self.assertEqual(44, d[idna_encode(self.utf8)])
+
+        # setting with idna version replaces something set previously with utf8
+        d[idna_encode(self.almost)] = 45
+        self.assertEqual(45, d[self.almost])
+        self.assertEqual(45, d[idna_encode(self.almost)])
+
+        # contains
+        self.assertTrue(self.plain in d)
+        self.assertTrue(self.almost in d)
+        self.assertTrue(idna_encode(self.almost) in d)
+        self.assertTrue(self.utf8 in d)
+        self.assertTrue(idna_encode(self.utf8) in d)
+
+        # we can delete with either form
+        del d[self.almost]
+        self.assertFalse(self.almost in d)
+        self.assertFalse(idna_encode(self.almost) in d)
+        del d[idna_encode(self.utf8)]
+        self.assertFalse(self.utf8 in d)
+        self.assertFalse(idna_encode(self.utf8) in d)
+
+        # smoke test of repr
+        d.__repr__()
+
+    def test_keys(self):
+        d = IdnaDict(self.normal)
+
+        # keys are idna versions by default
+        self.assertEqual(
+            (self.plain, idna_encode(self.almost), idna_encode(self.utf8)),
+            tuple(d.keys()),
+        )
+
+        # decoded keys gives the utf8 version
+        self.assertEqual(
+            (self.plain, self.almost, self.utf8), tuple(d.decoded_keys())
+        )
+
+    def test_items(self):
+        d = IdnaDict(self.normal)
+
+        # idna keys in items
+        self.assertEqual(
+            (
+                (self.plain, 42),
+                (idna_encode(self.almost), 43),
+                (idna_encode(self.utf8), 44),
+            ),
+            tuple(d.items()),
+        )
+
+        # utf8 keys in decoded_items
+        self.assertEqual(
+            ((self.plain, 42), (self.almost, 43), (self.utf8, 44)),
+            tuple(d.decoded_items()),
+        )
diff --git a/tests/test_octodns_manager.py b/tests/test_octodns_manager.py
index 7ea9202..c0cbfca 100644
--- a/tests/test_octodns_manager.py
+++ b/tests/test_octodns_manager.py
@@ -13,6 +13,7 @@ from os import environ
 from os.path import dirname, isfile, join
 
 from octodns import __VERSION__
+from octodns.idna import IdnaDict, idna_encode
 from octodns.manager import (
     _AggregateTarget,
     MainThreadExecutor,
@@ -182,6 +183,50 @@ class TestManager(TestCase):
             ).sync(dry_run=False, force=True)
             self.assertEqual(33, tc)
 
+    def test_idna_eligible_zones(self):
+        # loading w/simple, but we'll be blowing it away and doing some manual
+        # stuff
+        manager = Manager(get_config_filename('simple.yaml'))
+
+        # these configs won't be valid, but that's fine we can test what we're
+        # after based on exceptions raised
+        manager.config['zones'] = manager._config_zones(
+            {'déjà.vu.': {}, 'deja.vu.': {}, idna_encode('こんにちは.jp.'): {}}
+        )
+        from pprint import pprint
+
+        pprint(manager.config['zones'])
+
+        # refer to them with utf-8
+        with self.assertRaises(ManagerException) as ctx:
+            manager.sync(eligible_zones=('déjà.vu.',))
+        self.assertEqual('Zone déjà.vu. is missing sources', str(ctx.exception))
+
+        with self.assertRaises(ManagerException) as ctx:
+            manager.sync(eligible_zones=('deja.vu.',))
+        self.assertEqual('Zone deja.vu. is missing sources', str(ctx.exception))
+
+        with self.assertRaises(ManagerException) as ctx:
+            manager.sync(eligible_zones=('こんにちは.jp.',))
+        self.assertEqual(
+            'Zone こんにちは.jp. is missing sources', str(ctx.exception)
+        )
+
+        # refer to them with idna (exceptions are still utf-8
+        with self.assertRaises(ManagerException) as ctx:
+            manager.sync(eligible_zones=(idna_encode('déjà.vu.'),))
+        self.assertEqual('Zone déjà.vu. is missing sources', str(ctx.exception))
+
+        with self.assertRaises(ManagerException) as ctx:
+            manager.sync(eligible_zones=(idna_encode('deja.vu.'),))
+        self.assertEqual('Zone deja.vu. is missing sources', str(ctx.exception))
+
+        with self.assertRaises(ManagerException) as ctx:
+            manager.sync(eligible_zones=(idna_encode('こんにちは.jp.'),))
+        self.assertEqual(
+            'Zone こんにちは.jp. is missing sources', str(ctx.exception)
+        )
+
     def test_eligible_sources(self):
         with TemporaryDirectory() as tmpdir:
             environ['YAML_TMP_DIR'] = tmpdir.dirname
@@ -236,7 +281,7 @@ class TestManager(TestCase):
                     get_config_filename('simple-alias-zone.yaml')
                 ).sync(eligible_zones=["alias.tests."])
             self.assertEqual(
-                'Zone alias.tests. cannot be sync without zone '
+                'Zone alias.tests. cannot be synced without zone '
                 'unit.tests. sinced it is aliased',
                 str(ctx.exception),
             )
@@ -831,6 +876,41 @@ class TestManager(TestCase):
             set(), manager.configured_sub_zones('bar.foo.unit.tests.')
         )
 
+    def test_config_zones(self):
+        manager = Manager(get_config_filename('simple.yaml'))
+
+        # empty == empty
+        self.assertEqual({}, manager._config_zones({}))
+
+        # single ascii comes back as-is, but in a IdnaDict
+        zones = manager._config_zones({'unit.tests.': 42})
+        self.assertEqual({'unit.tests.': 42}, zones)
+        self.assertIsInstance(zones, IdnaDict)
+
+        # single utf-8 comes back idna encoded
+        self.assertEqual(
+            {idna_encode('Déjà.vu.'): 42},
+            dict(manager._config_zones({'Déjà.vu.': 42})),
+        )
+
+        # ascii and non-matching idna as ok
+        self.assertEqual(
+            {idna_encode('déjà.vu.'): 42, 'deja.vu.': 43},
+            dict(
+                manager._config_zones(
+                    {idna_encode('déjà.vu.'): 42, 'deja.vu.': 43}
+                )
+            ),
+        )
+
+        with self.assertRaises(ManagerException) as ctx:
+            # zone configured with both utf-8 and idna is an error
+            manager._config_zones({'Déjà.vu.': 42, idna_encode('Déjà.vu.'): 43})
+        self.assertEqual(
+            '"déjà.vu." configured both in utf-8 and idna "xn--dj-kia8a.vu."',
+            str(ctx.exception),
+        )
+
 
 class TestMainThreadExecutor(TestCase):
     def test_success(self):
diff --git a/tests/test_octodns_processor_filter.py b/tests/test_octodns_processor_filter.py
index 859677d..aa6f5ff 100644
--- a/tests/test_octodns_processor_filter.py
+++ b/tests/test_octodns_processor_filter.py
@@ -11,7 +11,12 @@ from __future__ import (
 
 from unittest import TestCase
 
-from octodns.processor.filter import TypeAllowlistFilter, TypeRejectlistFilter
+from octodns.processor.filter import (
+    NameAllowlistFilter,
+    NameRejectlistFilter,
+    TypeAllowlistFilter,
+    TypeRejectlistFilter,
+)
 from octodns.record import Record
 from octodns.zone import Zone
 
@@ -76,3 +81,83 @@ class TestTypeRejectListFilter(TestCase):
         filter_a_aaaa = TypeRejectlistFilter('not-a-aaaa', set(('A', 'AAAA')))
         got = filter_a_aaaa.process_target_zone(zone.copy())
         self.assertEqual(['txt', 'txt2'], sorted([r.name for r in got.records]))
+
+
+class TestNameAllowListFilter(TestCase):
+    zone = Zone('unit.tests.', [])
+    matches = Record.new(
+        zone, 'matches', {'type': 'A', 'ttl': 42, 'value': '1.2.3.4'}
+    )
+    zone.add_record(matches)
+    doesnt = Record.new(
+        zone, 'doesnt', {'type': 'A', 'ttl': 42, 'value': '2.3.4.5'}
+    )
+    zone.add_record(doesnt)
+    matchable1 = Record.new(
+        zone, 'start-f43ad96-end', {'type': 'A', 'ttl': 42, 'value': '3.4.5.6'}
+    )
+    zone.add_record(matchable1)
+    matchable2 = Record.new(
+        zone, 'start-a3b444c-end', {'type': 'A', 'ttl': 42, 'value': '4.5.6.7'}
+    )
+    zone.add_record(matchable2)
+
+    def test_exact(self):
+        allows = NameAllowlistFilter('exact', ('matches',))
+
+        self.assertEqual(4, len(self.zone.records))
+        filtered = allows.process_source_zone(self.zone.copy())
+        self.assertEqual(1, len(filtered.records))
+        self.assertEqual(['matches'], [r.name for r in filtered.records])
+
+    def test_regex(self):
+        allows = NameAllowlistFilter('exact', ('/^start-.+-end$/',))
+
+        self.assertEqual(4, len(self.zone.records))
+        filtered = allows.process_source_zone(self.zone.copy())
+        self.assertEqual(2, len(filtered.records))
+        self.assertEqual(
+            ['start-a3b444c-end', 'start-f43ad96-end'],
+            sorted([r.name for r in filtered.records]),
+        )
+
+
+class TestNameRejectListFilter(TestCase):
+    zone = Zone('unit.tests.', [])
+    matches = Record.new(
+        zone, 'matches', {'type': 'A', 'ttl': 42, 'value': '1.2.3.4'}
+    )
+    zone.add_record(matches)
+    doesnt = Record.new(
+        zone, 'doesnt', {'type': 'A', 'ttl': 42, 'value': '2.3.4.5'}
+    )
+    zone.add_record(doesnt)
+    matchable1 = Record.new(
+        zone, 'start-f43ad96-end', {'type': 'A', 'ttl': 42, 'value': '3.4.5.6'}
+    )
+    zone.add_record(matchable1)
+    matchable2 = Record.new(
+        zone, 'start-a3b444c-end', {'type': 'A', 'ttl': 42, 'value': '4.5.6.7'}
+    )
+    zone.add_record(matchable2)
+
+    def test_exact(self):
+        rejects = NameRejectlistFilter('exact', ('matches',))
+
+        self.assertEqual(4, len(self.zone.records))
+        filtered = rejects.process_source_zone(self.zone.copy())
+        self.assertEqual(3, len(filtered.records))
+        self.assertEqual(
+            ['doesnt', 'start-a3b444c-end', 'start-f43ad96-end'],
+            sorted([r.name for r in filtered.records]),
+        )
+
+    def test_regex(self):
+        rejects = NameRejectlistFilter('exact', ('/^start-.+-end$/',))
+
+        self.assertEqual(4, len(self.zone.records))
+        filtered = rejects.process_source_zone(self.zone.copy())
+        self.assertEqual(2, len(filtered.records))
+        self.assertEqual(
+            ['doesnt', 'matches'], sorted([r.name for r in filtered.records])
+        )
diff --git a/tests/test_octodns_processor_restrict.py b/tests/test_octodns_processor_restrict.py
new file mode 100644
index 0000000..4848ae6
--- /dev/null
+++ b/tests/test_octodns_processor_restrict.py
@@ -0,0 +1,113 @@
+from unittest import TestCase
+
+from octodns.processor.restrict import (
+    RestrictionException,
+    TtlRestrictionFilter,
+)
+from octodns.record import Record
+from octodns.zone import Zone
+
+
+class TestTtlRestrictionFilter(TestCase):
+    def test_restrict_ttl(self):
+        # configured values
+        restrictor = TtlRestrictionFilter('test', min_ttl=32, max_ttl=1024)
+
+        zone = Zone('unit.tests.', [])
+        good = Record.new(
+            zone, 'good', {'type': 'A', 'ttl': 42, 'value': '1.2.3.4'}
+        )
+        zone.add_record(good)
+
+        restricted = restrictor.process_source_zone(zone)
+        self.assertEqual(zone.records, restricted.records)
+
+        # too low
+        low = Record.new(
+            zone, 'low', {'type': 'A', 'ttl': 16, 'value': '1.2.3.4'}
+        )
+        copy = zone.copy()
+        copy.add_record(low)
+        with self.assertRaises(RestrictionException) as ctx:
+            restrictor.process_source_zone(copy)
+        self.assertEqual(
+            'low.unit.tests. ttl=16 too low, min_ttl=32', str(ctx.exception)
+        )
+
+        # with lenient set, we can go lower
+        lenient = Record.new(
+            zone,
+            'low',
+            {
+                'octodns': {'lenient': True},
+                'type': 'A',
+                'ttl': 16,
+                'value': '1.2.3.4',
+            },
+        )
+        copy = zone.copy()
+        copy.add_record(lenient)
+        restricted = restrictor.process_source_zone(copy)
+        self.assertEqual(copy.records, restricted.records)
+
+        # too high
+        high = Record.new(
+            zone, 'high', {'type': 'A', 'ttl': 2048, 'value': '1.2.3.4'}
+        )
+        copy = zone.copy()
+        copy.add_record(high)
+        with self.assertRaises(RestrictionException) as ctx:
+            restrictor.process_source_zone(copy)
+        self.assertEqual(
+            'high.unit.tests. ttl=2048 too high, max_ttl=1024',
+            str(ctx.exception),
+        )
+
+        # too low defaults
+        restrictor = TtlRestrictionFilter('test')
+        low = Record.new(
+            zone, 'low', {'type': 'A', 'ttl': 0, 'value': '1.2.3.4'}
+        )
+        copy = zone.copy()
+        copy.add_record(low)
+        with self.assertRaises(RestrictionException) as ctx:
+            restrictor.process_source_zone(copy)
+        self.assertEqual(
+            'low.unit.tests. ttl=0 too low, min_ttl=1', str(ctx.exception)
+        )
+
+        # too high defaults
+        high = Record.new(
+            zone, 'high', {'type': 'A', 'ttl': 999999, 'value': '1.2.3.4'}
+        )
+        copy = zone.copy()
+        copy.add_record(high)
+        with self.assertRaises(RestrictionException) as ctx:
+            restrictor.process_source_zone(copy)
+        self.assertEqual(
+            'high.unit.tests. ttl=999999 too high, max_ttl=604800',
+            str(ctx.exception),
+        )
+
+        # allowed_ttls
+        restrictor = TtlRestrictionFilter('test', allowed_ttls=[42, 300])
+
+        # add 300 (42 is already there)
+        another = Record.new(
+            zone, 'another', {'type': 'A', 'ttl': 300, 'value': '4.5.6.7'}
+        )
+        zone.add_record(another)
+
+        # 42 and 300 are allowed through
+        restricted = restrictor.process_source_zone(zone)
+        self.assertEqual(zone.records, restricted.records)
+
+        # 16 is not
+        copy = zone.copy()
+        copy.add_record(low)
+        with self.assertRaises(RestrictionException) as ctx:
+            restrictor.process_source_zone(copy)
+        self.assertEqual(
+            'low.unit.tests. ttl=0 not an allowed value, allowed_ttls={42, 300}',
+            str(ctx.exception),
+        )
diff --git a/tests/test_octodns_provider_yaml.py b/tests/test_octodns_provider_yaml.py
index 92939aa..f5b823f 100644
--- a/tests/test_octodns_provider_yaml.py
+++ b/tests/test_octodns_provider_yaml.py
@@ -15,7 +15,9 @@ from unittest import TestCase
 from yaml import safe_load
 from yaml.constructor import ConstructorError
 
+from octodns.idna import idna_encode
 from octodns.record import _NsValue, Create, Record, ValuesMixin
+from octodns.provider import ProviderException
 from octodns.provider.base import Plan
 from octodns.provider.yaml import (
     _list_all_yaml_files,
@@ -172,6 +174,58 @@ class TestYamlProvider(TestCase):
                 # make sure nothing is left
                 self.assertEqual([], list(data.keys()))
 
+    def test_idna(self):
+        with TemporaryDirectory() as td:
+            name = 'déjà.vu.'
+            filename = f'{name}yaml'
+
+            provider = YamlProvider('test', td.dirname)
+            zone = Zone(idna_encode(name), [])
+
+            # create a idna named file
+            with open(join(td.dirname, idna_encode(filename)), 'w') as fh:
+                fh.write(
+                    '''---
+'':
+  type: A
+  value: 1.2.3.4
+# something in idna notation
+xn--dj-kia8a:
+  type: A
+  value: 2.3.4.5
+# something with utf-8
+これはテストです:
+  type: A
+  value: 3.4.5.6
+'''
+                )
+
+            # populates fine when there's just the idna version (as a fallback)
+            provider.populate(zone)
+            d = {r.name: r for r in zone.records}
+            self.assertEqual(3, len(d))
+            # verify that we loaded the expected records, including idna/utf-8
+            # named ones
+            self.assertEqual(['1.2.3.4'], d[''].values)
+            self.assertEqual(['2.3.4.5'], d['xn--dj-kia8a'].values)
+            self.assertEqual(['3.4.5.6'], d['xn--28jm5b5a8k5k8cra'].values)
+
+            # create a utf8 named file (provider always writes utf-8 filenames
+            plan = provider.plan(zone)
+            provider.apply(plan)
+
+            with open(join(td.dirname, filename), 'r') as fh:
+                content = fh.read()
+                # verify that the non-ascii records were written out in utf-8
+                self.assertTrue('déjà:' in content)
+                self.assertTrue('これはテストです:' in content)
+
+            # does not allow both idna and utf8 named files
+            with self.assertRaises(ProviderException) as ctx:
+                provider.populate(zone)
+            msg = str(ctx.exception)
+            self.assertTrue('Both UTF-8' in msg)
+
     def test_empty(self):
         source = YamlProvider(
             'test', join(dirname(__file__), 'config'), supports_root_ns=False
diff --git a/tests/test_octodns_record.py b/tests/test_octodns_record.py
index 5521584..37c0180 100644
--- a/tests/test_octodns_record.py
+++ b/tests/test_octodns_record.py
@@ -11,6 +11,7 @@ from __future__ import (
 
 from unittest import TestCase
 
+from octodns.idna import idna_encode
 from octodns.record import (
     ARecord,
     AaaaRecord,
@@ -88,6 +89,18 @@ class TestRecord(TestCase):
         )
         self.assertEqual('mixedcase', record.name)
 
+    def test_utf8(self):
+        zone = Zone('natación.mx.', [])
+        utf8 = 'niño'
+        encoded = idna_encode(utf8)
+        record = ARecord(
+            zone, utf8, {'ttl': 30, 'type': 'A', 'value': '1.2.3.4'}
+        )
+        self.assertEqual(encoded, record.name)
+        self.assertEqual(utf8, record.decoded_name)
+        self.assertTrue(f'{encoded}.{zone.name}', record.fqdn)
+        self.assertTrue(f'{utf8}.{zone.decoded_name}', record.decoded_fqdn)
+
     def test_alias_lowering_value(self):
         upper_record = AliasRecord(
             self.zone,
@@ -2013,6 +2026,51 @@ class TestRecordValidation(TestCase):
             self.zone, name, {'ttl': 300, 'type': 'A', 'value': '1.2.3.4'}
         )
 
+        # make sure we're validating with encoded fqdns
+        utf8 = 'déjà-vu'
+        padding = ('.' + ('x' * 57)) * 4
+        utf8_name = f'{utf8}{padding}'
+        # make sure our test is valid here, we're under 253 chars long as utf8
+        self.assertEqual(251, len(f'{utf8_name}.{self.zone.name}'))
+        with self.assertRaises(ValidationError) as ctx:
+            Record.new(
+                self.zone,
+                utf8_name,
+                {'ttl': 300, 'type': 'A', 'value': '1.2.3.4'},
+            )
+        reason = ctx.exception.reasons[0]
+        self.assertTrue(reason.startswith('invalid fqdn, "déjà-vu'))
+        self.assertTrue(
+            reason.endswith(
+                '.unit.tests." is too long at 259' ' chars, max is 253'
+            )
+        )
+
+        # same, but with ascii version of things
+        plain = 'deja-vu'
+        plain_name = f'{plain}{padding}'
+        self.assertEqual(251, len(f'{plain_name}.{self.zone.name}'))
+        Record.new(
+            self.zone, plain_name, {'ttl': 300, 'type': 'A', 'value': '1.2.3.4'}
+        )
+
+        # check that we're validating encoded labels
+        padding = 'x' * (60 - len(utf8))
+        utf8_name = f'{utf8}{padding}'
+        # make sure the test is valid, we're at 63 chars
+        self.assertEqual(60, len(utf8_name))
+        with self.assertRaises(ValidationError) as ctx:
+            Record.new(
+                self.zone,
+                utf8_name,
+                {'ttl': 300, 'type': 'A', 'value': '1.2.3.4'},
+            )
+        reason = ctx.exception.reasons[0]
+        # Unfortunately this is a translated IDNAError so we don't have much
+        # control over the exact message :-/ (doesn't give context like octoDNS
+        # does)
+        self.assertEqual('Label too long', reason)
+
         # no ttl
         with self.assertRaises(ValidationError) as ctx:
             Record.new(self.zone, '', {'type': 'A', 'value': '1.2.3.4'})
diff --git a/tests/test_octodns_zone.py b/tests/test_octodns_zone.py
index 37e893f..5245735 100644
--- a/tests/test_octodns_zone.py
+++ b/tests/test_octodns_zone.py
@@ -11,6 +11,7 @@ from __future__ import (
 
 from unittest import TestCase
 
+from octodns.idna import idna_encode
 from octodns.record import (
     ARecord,
     AaaaRecord,
@@ -35,6 +36,13 @@ class TestZone(TestCase):
         zone = Zone('UniT.TEsTs.', [])
         self.assertEqual('unit.tests.', zone.name)
 
+    def test_utf8(self):
+        utf8 = 'grüßen.de.'
+        encoded = idna_encode(utf8)
+        zone = Zone(utf8, [])
+        self.assertEqual(encoded, zone.name)
+        self.assertEqual(utf8, zone.decoded_name)
+
     def test_hostname_from_fqdn(self):
         zone = Zone('unit.tests.', [])
         for hostname, fqdn in (
@@ -46,6 +54,27 @@ class TestZone(TestCase):
             ('foo.bar', 'foo.bar.unit.tests'),
             ('foo.unit.tests', 'foo.unit.tests.unit.tests.'),
             ('foo.unit.tests', 'foo.unit.tests.unit.tests'),
+            ('déjà', 'déjà.unit.tests'),
+            ('déjà.foo', 'déjà.foo.unit.tests'),
+            ('bar.déjà', 'bar.déjà.unit.tests'),
+            ('bar.déjà.foo', 'bar.déjà.foo.unit.tests'),
+        ):
+            self.assertEqual(hostname, zone.hostname_from_fqdn(fqdn))
+
+        zone = Zone('grüßen.de.', [])
+        for hostname, fqdn in (
+            ('', 'grüßen.de.'),
+            ('', 'grüßen.de'),
+            ('foo', 'foo.grüßen.de.'),
+            ('foo', 'foo.grüßen.de'),
+            ('foo.bar', 'foo.bar.grüßen.de.'),
+            ('foo.bar', 'foo.bar.grüßen.de'),
+            ('foo.grüßen.de', 'foo.grüßen.de.grüßen.de.'),
+            ('foo.grüßen.de', 'foo.grüßen.de.grüßen.de'),
+            ('déjà', 'déjà.grüßen.de'),
+            ('déjà.foo', 'déjà.foo.grüßen.de'),
+            ('bar.déjà', 'bar.déjà.grüßen.de'),
+            ('bar.déjà.foo', 'bar.déjà.foo.grüßen.de'),
         ):
             self.assertEqual(hostname, zone.hostname_from_fqdn(fqdn))