cloudfloordns.client.sync.zone
1# from dataclasses import dataclass, field 2import logging 3from typing import List, Optional, Union 4 5from cloudfloordns.models import Redirect, Zone 6 7from .pool import POOL 8 9DEFAULT_PRIMARY_NS = "ns1.g02.cfdns.net" 10 11 12class Zones: 13 def __init__(self, client) -> None: 14 self._client = client 15 16 # def create(self, domain: str, record: Record, timeout=None): 17 # url = f"/dns/zone/{domain}/record" 18 # return self._client.post( 19 # url, 20 # data=record.model_dump(), 21 # timeout=timeout, 22 # ) 23 24 def update(self, domain: "Zone", soa=None, timeout=None): 25 url = "/dns/zone" 26 data = domain.model_dump(exclude_unset=True) 27 if not soa: 28 soa = self.soa(domain) 29 30 data = domain.model_dump(exclude_unset=True) 31 soa_data = { 32 "master": soa["ns"], # NS: primary name server 33 "retry": soa[ 34 "retry" 35 ], # Retry: How often secondaries attempt to fetch the zone if the first refresh fails 36 "refresh": soa[ 37 "refresh" 38 ], # Refresh: How often secondaries should check if changes are made to the zone 39 "expire": soa[ 40 "expire" 41 ], # Expire: Secondaries will discard the zone if no refresh could be made within this interval. 42 "min": soa[ 43 "minimum" 44 ], # Min TTL: default TTL for new records. Also determines how long negative records are cached (record not found) 45 "mbox": soa[ 46 "mbox" 47 ], # RP: Responsible person (email address with period instead of '@') 48 "ttl": soa[ 49 "ttl" 50 ], # SOA TTL: Number of seconds this zone may be cached before the source must be consulted again. 51 } 52 request_data = {**soa_data, **data} 53 return self._client.patch( 54 url, 55 data=request_data, 56 timeout=timeout, 57 ) 58 59 def disable(self, domain: str, timeout=None): 60 url = f"/dns/zone/{domain}" 61 return self._client.delete( 62 url, 63 timeout=timeout, 64 ) 65 66 def enable( 67 self, 68 domain, 69 master=DEFAULT_PRIMARY_NS, 70 # master="dns0.mtgsy.co.uk.", 71 retry=1200, 72 refresh=3600, 73 expire=1209600, 74 min=3600, 75 responsible="hostmaster", 76 ttl=86400, 77 timeout=None, 78 ): 79 if not isinstance(domain, str): 80 domain = domain.domainname 81 url = f"/dns/zone/{domain}/enable" 82 # This will create the SOA record 83 # The default values can be found on an active domain 84 return self._client.patch( 85 url, 86 data={ 87 "domainname": domain, 88 "master": master, # NS: primary name server 89 "retry": retry, # Retry: How often secondaries attempt to fetch the zone if the first refresh fails 90 "refresh": refresh, # Refresh: How often secondaries should check if changes are made to the zone 91 "expire": expire, # Expire: Secondaries will discard the zone if no refresh could be made within this interval. 92 "min": min, # Min TTL: default TTL for new records. Also determines how long negative records are cached (record not found) 93 "mbox": responsible, # RP: Responsible person (email address with period instead of '@') 94 "ttl": ttl, # SOA TTL: Number of seconds this zone may be cached before the source must be consulted again. 95 }, 96 timeout=timeout, 97 ) 98 99 def enable_all( 100 self, 101 master=DEFAULT_PRIMARY_NS, 102 # master="dns0.mtgsy.co.uk.", 103 retry=1200, 104 refresh=3600, 105 expire=1209600, 106 min=3600, 107 responsible="hostmaster", 108 ttl=86400, 109 timeout=None, 110 ): 111 zones = {z.domainname for z in self.list()} 112 domains = [d for d in self._client.domains.list() if d.domainname not in zones] 113 114 # WORKAROUND: Get accurate data per domains, 115 # the nameservers returned by self._client.domains.list() are wrong 116 domains = (self._client.domains.get(d.domainname) for d in domains) 117 # Don't activate the one that are externally managed 118 domains = [d for d in domains if d.nameserver and not d.is_externally_managed] 119 120 def worker(domain): 121 try: 122 return domain.domainname, self._client.zones.enable( 123 domain, 124 master=master, 125 retry=retry, 126 refresh=refresh, 127 expire=expire, 128 min=min, 129 responsible=responsible, 130 ttl=ttl, 131 timeout=timeout, 132 ) 133 except Exception as e: 134 return domain.domainname, str(e) 135 136 return POOL.map(worker, domains) 137 138 def list(self, timeout=None) -> List[Zone]: 139 url = "/dns/zone" 140 res = self._client.get( 141 url, 142 timeout=timeout, 143 ) 144 return [Zone.model_validate(d) for d in res] 145 146 def raw_list_redirects(self, zone: str, timeout=None) -> List[dict]: 147 url = f"/domain/{zone}/get_domain_forward" 148 try: 149 return self._client.get( 150 url, 151 timeout=timeout, 152 ) 153 except Exception as e: 154 if "No Domain forward available for requested domain." in str(e): 155 logging.debug(f"Zone {zone} has no redirect target") 156 return [] 157 raise 158 159 def list_redirects(self, zone: str, timeout=None) -> List[Redirect]: 160 return [ 161 Redirect.model_validate(r) 162 for r in self.raw_list_redirects(zone, timeout=timeout) 163 ] 164 165 def get(self, domain_id: str, zone_enabled: Optional[bool] = None, timeout=None): 166 res = self.list( 167 timeout=timeout, 168 ) 169 return next((r for r in res if r.id == domain_id), None) 170 171 def get_by_name( 172 self, domainname: str, zone_enabled: Optional[bool] = None, timeout=None 173 ): 174 res = self.list( 175 timeout=timeout, 176 ) 177 return next((r for r in res if r.domainname == domainname), None) 178 179 def soa(self, domain: Union[str, Zone], timeout=None): 180 if isinstance(domain, Zone): 181 domain = domain.domainname 182 url = f"/dns/zone/{domain}/soa" 183 return self._client.get( 184 url, 185 timeout=timeout, 186 ) 187 188 def update_soa( 189 self, 190 domain, 191 master=None, 192 serial=None, 193 retry=None, 194 refresh=None, 195 expire=None, 196 min=None, 197 responsible=None, 198 ttl=None, 199 xfer=None, 200 timeout=None, 201 ): 202 if not isinstance(domain, str): 203 domain = domain.domainname 204 url = f"/dns/zone/{domain}/soa" 205 data = { 206 "ns": master, # NS: primary name server 207 "retry": retry, # Retry: How often secondaries attempt to fetch the zone if the first refresh fails 208 "refresh": refresh, # Refresh: How often secondaries should check if changes are made to the zone 209 "expire": expire, # Expire: Secondaries will discard the zone if no refresh could be made within this interval. 210 "minimum": min, # Min TTL: default TTL for new records. Also determines how long negative records are cached (record not found) 211 "mbox": responsible, # RP: Responsible person (email address with period instead of '@') 212 "ttl": ttl, # SOA TTL: Number of seconds this zone may be cached before the source must be consulted again. 213 "serial": serial, 214 "xfer": xfer, 215 } 216 data = {k: v for k, v in data.items() if v is not None} 217 return self._client.patch( 218 url, 219 data=data, 220 timeout=timeout, 221 ) 222 223 224__all__ = [ 225 "Zones", 226]
class
Zones:
13class Zones: 14 def __init__(self, client) -> None: 15 self._client = client 16 17 # def create(self, domain: str, record: Record, timeout=None): 18 # url = f"/dns/zone/{domain}/record" 19 # return self._client.post( 20 # url, 21 # data=record.model_dump(), 22 # timeout=timeout, 23 # ) 24 25 def update(self, domain: "Zone", soa=None, timeout=None): 26 url = "/dns/zone" 27 data = domain.model_dump(exclude_unset=True) 28 if not soa: 29 soa = self.soa(domain) 30 31 data = domain.model_dump(exclude_unset=True) 32 soa_data = { 33 "master": soa["ns"], # NS: primary name server 34 "retry": soa[ 35 "retry" 36 ], # Retry: How often secondaries attempt to fetch the zone if the first refresh fails 37 "refresh": soa[ 38 "refresh" 39 ], # Refresh: How often secondaries should check if changes are made to the zone 40 "expire": soa[ 41 "expire" 42 ], # Expire: Secondaries will discard the zone if no refresh could be made within this interval. 43 "min": soa[ 44 "minimum" 45 ], # Min TTL: default TTL for new records. Also determines how long negative records are cached (record not found) 46 "mbox": soa[ 47 "mbox" 48 ], # RP: Responsible person (email address with period instead of '@') 49 "ttl": soa[ 50 "ttl" 51 ], # SOA TTL: Number of seconds this zone may be cached before the source must be consulted again. 52 } 53 request_data = {**soa_data, **data} 54 return self._client.patch( 55 url, 56 data=request_data, 57 timeout=timeout, 58 ) 59 60 def disable(self, domain: str, timeout=None): 61 url = f"/dns/zone/{domain}" 62 return self._client.delete( 63 url, 64 timeout=timeout, 65 ) 66 67 def enable( 68 self, 69 domain, 70 master=DEFAULT_PRIMARY_NS, 71 # master="dns0.mtgsy.co.uk.", 72 retry=1200, 73 refresh=3600, 74 expire=1209600, 75 min=3600, 76 responsible="hostmaster", 77 ttl=86400, 78 timeout=None, 79 ): 80 if not isinstance(domain, str): 81 domain = domain.domainname 82 url = f"/dns/zone/{domain}/enable" 83 # This will create the SOA record 84 # The default values can be found on an active domain 85 return self._client.patch( 86 url, 87 data={ 88 "domainname": domain, 89 "master": master, # NS: primary name server 90 "retry": retry, # Retry: How often secondaries attempt to fetch the zone if the first refresh fails 91 "refresh": refresh, # Refresh: How often secondaries should check if changes are made to the zone 92 "expire": expire, # Expire: Secondaries will discard the zone if no refresh could be made within this interval. 93 "min": min, # Min TTL: default TTL for new records. Also determines how long negative records are cached (record not found) 94 "mbox": responsible, # RP: Responsible person (email address with period instead of '@') 95 "ttl": ttl, # SOA TTL: Number of seconds this zone may be cached before the source must be consulted again. 96 }, 97 timeout=timeout, 98 ) 99 100 def enable_all( 101 self, 102 master=DEFAULT_PRIMARY_NS, 103 # master="dns0.mtgsy.co.uk.", 104 retry=1200, 105 refresh=3600, 106 expire=1209600, 107 min=3600, 108 responsible="hostmaster", 109 ttl=86400, 110 timeout=None, 111 ): 112 zones = {z.domainname for z in self.list()} 113 domains = [d for d in self._client.domains.list() if d.domainname not in zones] 114 115 # WORKAROUND: Get accurate data per domains, 116 # the nameservers returned by self._client.domains.list() are wrong 117 domains = (self._client.domains.get(d.domainname) for d in domains) 118 # Don't activate the one that are externally managed 119 domains = [d for d in domains if d.nameserver and not d.is_externally_managed] 120 121 def worker(domain): 122 try: 123 return domain.domainname, self._client.zones.enable( 124 domain, 125 master=master, 126 retry=retry, 127 refresh=refresh, 128 expire=expire, 129 min=min, 130 responsible=responsible, 131 ttl=ttl, 132 timeout=timeout, 133 ) 134 except Exception as e: 135 return domain.domainname, str(e) 136 137 return POOL.map(worker, domains) 138 139 def list(self, timeout=None) -> List[Zone]: 140 url = "/dns/zone" 141 res = self._client.get( 142 url, 143 timeout=timeout, 144 ) 145 return [Zone.model_validate(d) for d in res] 146 147 def raw_list_redirects(self, zone: str, timeout=None) -> List[dict]: 148 url = f"/domain/{zone}/get_domain_forward" 149 try: 150 return self._client.get( 151 url, 152 timeout=timeout, 153 ) 154 except Exception as e: 155 if "No Domain forward available for requested domain." in str(e): 156 logging.debug(f"Zone {zone} has no redirect target") 157 return [] 158 raise 159 160 def list_redirects(self, zone: str, timeout=None) -> List[Redirect]: 161 return [ 162 Redirect.model_validate(r) 163 for r in self.raw_list_redirects(zone, timeout=timeout) 164 ] 165 166 def get(self, domain_id: str, zone_enabled: Optional[bool] = None, timeout=None): 167 res = self.list( 168 timeout=timeout, 169 ) 170 return next((r for r in res if r.id == domain_id), None) 171 172 def get_by_name( 173 self, domainname: str, zone_enabled: Optional[bool] = None, timeout=None 174 ): 175 res = self.list( 176 timeout=timeout, 177 ) 178 return next((r for r in res if r.domainname == domainname), None) 179 180 def soa(self, domain: Union[str, Zone], timeout=None): 181 if isinstance(domain, Zone): 182 domain = domain.domainname 183 url = f"/dns/zone/{domain}/soa" 184 return self._client.get( 185 url, 186 timeout=timeout, 187 ) 188 189 def update_soa( 190 self, 191 domain, 192 master=None, 193 serial=None, 194 retry=None, 195 refresh=None, 196 expire=None, 197 min=None, 198 responsible=None, 199 ttl=None, 200 xfer=None, 201 timeout=None, 202 ): 203 if not isinstance(domain, str): 204 domain = domain.domainname 205 url = f"/dns/zone/{domain}/soa" 206 data = { 207 "ns": master, # NS: primary name server 208 "retry": retry, # Retry: How often secondaries attempt to fetch the zone if the first refresh fails 209 "refresh": refresh, # Refresh: How often secondaries should check if changes are made to the zone 210 "expire": expire, # Expire: Secondaries will discard the zone if no refresh could be made within this interval. 211 "minimum": min, # Min TTL: default TTL for new records. Also determines how long negative records are cached (record not found) 212 "mbox": responsible, # RP: Responsible person (email address with period instead of '@') 213 "ttl": ttl, # SOA TTL: Number of seconds this zone may be cached before the source must be consulted again. 214 "serial": serial, 215 "xfer": xfer, 216 } 217 data = {k: v for k, v in data.items() if v is not None} 218 return self._client.patch( 219 url, 220 data=data, 221 timeout=timeout, 222 )
25 def update(self, domain: "Zone", soa=None, timeout=None): 26 url = "/dns/zone" 27 data = domain.model_dump(exclude_unset=True) 28 if not soa: 29 soa = self.soa(domain) 30 31 data = domain.model_dump(exclude_unset=True) 32 soa_data = { 33 "master": soa["ns"], # NS: primary name server 34 "retry": soa[ 35 "retry" 36 ], # Retry: How often secondaries attempt to fetch the zone if the first refresh fails 37 "refresh": soa[ 38 "refresh" 39 ], # Refresh: How often secondaries should check if changes are made to the zone 40 "expire": soa[ 41 "expire" 42 ], # Expire: Secondaries will discard the zone if no refresh could be made within this interval. 43 "min": soa[ 44 "minimum" 45 ], # Min TTL: default TTL for new records. Also determines how long negative records are cached (record not found) 46 "mbox": soa[ 47 "mbox" 48 ], # RP: Responsible person (email address with period instead of '@') 49 "ttl": soa[ 50 "ttl" 51 ], # SOA TTL: Number of seconds this zone may be cached before the source must be consulted again. 52 } 53 request_data = {**soa_data, **data} 54 return self._client.patch( 55 url, 56 data=request_data, 57 timeout=timeout, 58 )
def
enable( self, domain, master='ns1.g02.cfdns.net', retry=1200, refresh=3600, expire=1209600, min=3600, responsible='hostmaster', ttl=86400, timeout=None):
67 def enable( 68 self, 69 domain, 70 master=DEFAULT_PRIMARY_NS, 71 # master="dns0.mtgsy.co.uk.", 72 retry=1200, 73 refresh=3600, 74 expire=1209600, 75 min=3600, 76 responsible="hostmaster", 77 ttl=86400, 78 timeout=None, 79 ): 80 if not isinstance(domain, str): 81 domain = domain.domainname 82 url = f"/dns/zone/{domain}/enable" 83 # This will create the SOA record 84 # The default values can be found on an active domain 85 return self._client.patch( 86 url, 87 data={ 88 "domainname": domain, 89 "master": master, # NS: primary name server 90 "retry": retry, # Retry: How often secondaries attempt to fetch the zone if the first refresh fails 91 "refresh": refresh, # Refresh: How often secondaries should check if changes are made to the zone 92 "expire": expire, # Expire: Secondaries will discard the zone if no refresh could be made within this interval. 93 "min": min, # Min TTL: default TTL for new records. Also determines how long negative records are cached (record not found) 94 "mbox": responsible, # RP: Responsible person (email address with period instead of '@') 95 "ttl": ttl, # SOA TTL: Number of seconds this zone may be cached before the source must be consulted again. 96 }, 97 timeout=timeout, 98 )
def
enable_all( self, master='ns1.g02.cfdns.net', retry=1200, refresh=3600, expire=1209600, min=3600, responsible='hostmaster', ttl=86400, timeout=None):
100 def enable_all( 101 self, 102 master=DEFAULT_PRIMARY_NS, 103 # master="dns0.mtgsy.co.uk.", 104 retry=1200, 105 refresh=3600, 106 expire=1209600, 107 min=3600, 108 responsible="hostmaster", 109 ttl=86400, 110 timeout=None, 111 ): 112 zones = {z.domainname for z in self.list()} 113 domains = [d for d in self._client.domains.list() if d.domainname not in zones] 114 115 # WORKAROUND: Get accurate data per domains, 116 # the nameservers returned by self._client.domains.list() are wrong 117 domains = (self._client.domains.get(d.domainname) for d in domains) 118 # Don't activate the one that are externally managed 119 domains = [d for d in domains if d.nameserver and not d.is_externally_managed] 120 121 def worker(domain): 122 try: 123 return domain.domainname, self._client.zones.enable( 124 domain, 125 master=master, 126 retry=retry, 127 refresh=refresh, 128 expire=expire, 129 min=min, 130 responsible=responsible, 131 ttl=ttl, 132 timeout=timeout, 133 ) 134 except Exception as e: 135 return domain.domainname, str(e) 136 137 return POOL.map(worker, domains)
def
raw_list_redirects(self, zone: str, timeout=None) -> List[dict]:
147 def raw_list_redirects(self, zone: str, timeout=None) -> List[dict]: 148 url = f"/domain/{zone}/get_domain_forward" 149 try: 150 return self._client.get( 151 url, 152 timeout=timeout, 153 ) 154 except Exception as e: 155 if "No Domain forward available for requested domain." in str(e): 156 logging.debug(f"Zone {zone} has no redirect target") 157 return [] 158 raise
def
update_soa( self, domain, master=None, serial=None, retry=None, refresh=None, expire=None, min=None, responsible=None, ttl=None, xfer=None, timeout=None):
189 def update_soa( 190 self, 191 domain, 192 master=None, 193 serial=None, 194 retry=None, 195 refresh=None, 196 expire=None, 197 min=None, 198 responsible=None, 199 ttl=None, 200 xfer=None, 201 timeout=None, 202 ): 203 if not isinstance(domain, str): 204 domain = domain.domainname 205 url = f"/dns/zone/{domain}/soa" 206 data = { 207 "ns": master, # NS: primary name server 208 "retry": retry, # Retry: How often secondaries attempt to fetch the zone if the first refresh fails 209 "refresh": refresh, # Refresh: How often secondaries should check if changes are made to the zone 210 "expire": expire, # Expire: Secondaries will discard the zone if no refresh could be made within this interval. 211 "minimum": min, # Min TTL: default TTL for new records. Also determines how long negative records are cached (record not found) 212 "mbox": responsible, # RP: Responsible person (email address with period instead of '@') 213 "ttl": ttl, # SOA TTL: Number of seconds this zone may be cached before the source must be consulted again. 214 "serial": serial, 215 "xfer": xfer, 216 } 217 data = {k: v for k, v in data.items() if v is not None} 218 return self._client.patch( 219 url, 220 data=data, 221 timeout=timeout, 222 )