ascension.py (23261B)
1 #!/usr/bin/env python3 2 """ 3 This file is part of Ascension. 4 Copyright (C) 2018-2022 GNUnet e.V. 5 6 Ascension is free software: you can redistribute it and/or modify it 7 under the terms of the GNU Affero General Public License as published 8 by the Free Software Foundation, either version 3 of the License, 9 or (at your option) any later version. 10 11 Ascension is distributed in the hope that it will be useful, but 12 WITHOUT ANY WARRANTY; without even the implied warranty of 13 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU 14 Affero General Public License for more details. 15 16 You should have received a copy of the GNU Affero General Public License 17 along with this program. If not, see <http://www.gnu.org/licenses/>. 18 19 SPDX-License-Identifier: AGPL3.0-or-later 20 21 Author: rexxnor 22 """ 23 24 import argparse 25 import logging 26 import os 27 import time 28 import subprocess 29 import itertools 30 import multiprocessing as mp 31 32 import dns.rdatatype 33 import dns.zone 34 35 import ascension.util.argumentparser 36 import ascension.util.classes 37 import ascension.util.constants 38 import ascension.util.keyfile 39 import ascension.util.rest 40 import ascension.util.transformers 41 42 def work_slice(n,step,pp_setslice,gn_path): 43 # TODO Check path / exit code? Config could be generated. 44 worker_cfg = 'namestore-ascension-worker-' + str(n) + '.conf' 45 with open(worker_cfg, 'w') as f: 46 f.write('[namestore]\n') 47 f.write('DATABASE = postgres\n') 48 worker_sock = 'UNIXPATH = $GNUNET_USER_RUNTIME_DIR/gnunet-service-namestore-'+str(n)+'.sock' 49 f.write(worker_sock) 50 51 ns_svc_full_path = os.path.join(gn_path, 'gnunet', 'libexec', 'gnunet-service-namestore') 52 ns_svc_process = subprocess.Popen([ns_svc_full_path, '-c', worker_cfg]) 53 ns_process = subprocess.Popen(["gnunet-namestore", "-B", str(step), "-a", "-S", "-c", worker_cfg], stdin=subprocess.PIPE, text=True) 54 start = time.time() 55 i = 0 56 j = 0 57 psetcount = len(pp_setslice) 58 for name, payload in pp_setslice.items(): 59 # log if the rdataset is empty for some reason 60 i += 1 61 if not payload: 62 print("Empty Rdataset!") 63 continue 64 j += len(payload.data) 65 if (i % step) == 0: 66 print("Worker #%d: Adding record set %d/%d for a total of %d records\n"%(n,i,psetcount,j), end="") 67 ns_process.stdin.write(name + ":\n") 68 for r in payload.data: 69 flags = "[r{}]".format('p' if not r.is_private else '') 70 # FIXME we have many more flags. but probably not in our use 71 # case? We always have relative expirations, for example. 72 ns_process.stdin.write("{} {} {} {}\n".format(r.record_type, 73 r.relative_expiration, 74 flags, 75 r.value)) 76 ns_process.stdin.close() 77 ns_process.wait() 78 ns_svc_process.terminate() 79 os.remove(worker_cfg) 80 81 class Ascension(): 82 """ 83 Provides migration utilities for any given domain that supports zone transfer 84 """ 85 def __init__(self, args: argparse.Namespace): 86 """Constructor initializing all the classes and variables needed""" 87 # Logging 88 logging.basicConfig() 89 self.logger = logging.getLogger(__name__) 90 self.logger.setLevel(int(args.loglevel)) 91 domain = args.domain 92 93 # special case for root zone 94 if args.domain[-1] == '.' and len(args.domain) == 1: 95 domain = '@' 96 if args.domain[-1] == '.': 97 domain = domain[:-1] 98 99 self.rrsetcount = 0 100 self.subzonedict = {} 101 self.gnunet_prefix = args.gnunetprefix 102 self.num_workers = int(args.workers) 103 self.batch_size = int(args.batchsize) 104 105 self.session = ascension.util.rest.GNUnetRestSession() 106 self.gnszone = ascension.util.classes.GNSZone( 107 self.session, domain, args.public, args.ttl, self.logger 108 ) 109 self.dnszone = ascension.util.classes.DNSZone( 110 domain, args.nameserver, args.port, args.keyfile 111 ) 112 self.transformer = ascension.util.transformers.Transformer( 113 domain, 114 self.dnszone 115 ) 116 117 118 119 def add_records_to_gns(self) -> None: 120 """ 121 Extracts records from transferred zone and adds them to GNS 122 :raises AttributeError: When getting incomplete data 123 """ 124 self.logger.info("Starting to add records into GNS...") 125 self.rrsetcount = 0 126 127 # Defining worker 128 def worker(labelrecords): 129 label = "" 130 bestlabel = "" 131 domain = None 132 133 # break if taskqueue is empty 134 if not labelrecords: 135 return 136 137 record_data = ascension.util.classes.GNSRRecordSet( 138 record_name=label, 139 data=[] 140 ) 141 142 # execute thing to run on item 143 label, listofrdatasets = labelrecords 144 label = str(label) 145 146 subzones = str(label).split('.') 147 domain = self.gnszone.domain 148 bestlabel = label 149 150 if len(subzones) > 1: 151 label = subzones[0] 152 subdomains = ".".join(subzones[1:]) 153 subzone = f"{subdomains}.{domain}" 154 fqdn = f"{label}.{subdomains}.{domain}" 155 if subzone.startswith(("_tcp", "_udp")): 156 subzone = subzone.lstrip('_tcp.') 157 subzone = subzone.lstrip('_udp.') 158 bestlabel = label 159 if fqdn in self.subzonedict: 160 label = "@" 161 domain = fqdn 162 elif subzone in self.subzonedict: 163 if any(proto in fqdn for proto in ('_tcp', '_udp')): 164 fragment = fqdn.split('.') 165 bestlabel = '.'.join(fragment[0:2]) 166 domain = '.'.join(fragment[2:]) 167 else: 168 domain = subzone 169 170 for rdataset in listofrdatasets: 171 for record in rdataset: 172 rdtype = dns.rdatatype.to_text(record.rdtype) 173 if rdtype not in ascension.util.constants.PROCESSABLE_RECORD_TYPES: 174 self.logger.debug("%s records not supported!", rdtype) 175 continue 176 177 try: 178 if rdataset.ttl <= self.gnszone.minimum: 179 ttl = self.gnszone.minimum 180 else: 181 ttl = rdataset.ttl 182 except AttributeError: 183 ttl = self.gnszone.minimum 184 185 value = str(record) 186 187 # ignore NS for itself here 188 if label == '@' and rdtype == 'NS': 189 self.logger.debug("ignoring NS record for itself") 190 191 # modify value to fit gns syntax 192 rdtype, value, label = \ 193 self.transformer.transform_to_gns_format(record, 194 rdtype, 195 domain, 196 bestlabel) 197 # skip record if value is none 198 if value is None: 199 continue 200 201 # if label has changed, adjust GNSRecordData label as well 202 if record_data.record_name != label: 203 record_data.record_name = label 204 205 if isinstance(value, list): 206 for element in value: 207 entry = ascension.util.classes.GNSRecordData( 208 value=element, 209 record_type=rdtype, 210 relative_expiration=ttl, 211 is_relative_expiration=True, 212 is_private=not self.gnszone.public 213 ) 214 record_data.data.append(entry) 215 else: 216 entry = ascension.util.classes.GNSRecordData( 217 value=value, 218 record_type=rdtype, 219 relative_expiration=ttl, 220 is_relative_expiration=True, 221 is_private=not self.gnszone.public 222 ) 223 record_data.data.append(entry) 224 225 payload = record_data 226 if not record_data.data: 227 self.logger.warning("Empty record %s", record_data) 228 return "", None 229 #self.logger.debug("Payload: %s", payload.to_json()) 230 231 # Replace the records already present in GNS as old ones are not deleted 232 self.logger.debug(payload.record_name + "." + domain + ":\n") 233 return payload.record_name + "." + domain, payload 234 # FIXME error checking 235 #response = self.session.post(f"/namestore/{domain}", data=payload.to_json()) 236 237 #if response.status_code == 204: 238 # self.logger.debug("Record(s) with label %s added", label) 239 #else: 240 # data = response.json() 241 # error = data.get('error') 242 # self.logger.error("Unable to add record %s at URL %s: %s", 243 # record_data.to_json(), 244 # f"{self.session.base_url}/namestore/{domain}", 245 # error) 246 247 self.rrsetcount = self.rrsetcount + 1 248 # End of worker 249 250 # Building hierarchy afterwards 251 tstart = time.time() 252 self.create_zone_hierarchy() 253 # Needs to happen after the previous line and before the adding of records 254 self.transformer.subzonedict = self.subzonedict 255 tend = time.time() 256 self.logger.info("Zone hierarchy in %s seconds", str(tend - tstart)) 257 258 # Do it single threaded because threading scares me 259 setcount = len(self.dnszone.zone.nodes.items()) 260 pp_set = {} 261 rrcount = 0 262 # TODO: 263 # So, what we want to do here is to get all "dirty" 264 # record sets. Dirty record sets are records that were 265 # modified during the last pass (should here always be 266 # > 0 since serial changed) 267 # We may have just been given an AXFR even though IXFR was 268 # requested. 269 # So, after we add the records with the new serial, we 270 # should delete all records with the old. 271 # CAREFUL: This would mean that if the did receive an IXFR, 272 # we have to update all serial numbers in the DB! 273 for name, rdatasets in self.dnszone.zone.nodes.items(): 274 # log if the rdataset is empty for some reason 275 if not rdatasets: 276 print("Empty Rdataset!") 277 continue 278 name,payload = worker((name, rdatasets)) 279 if payload == None: 280 continue 281 pp_set[name] = payload 282 if payload: 283 rrcount += len(payload.data) 284 285 pp_setcount = len(pp_set.items()) 286 left = pp_setcount 287 start = 0 288 slice0count = int(pp_setcount/self.num_workers) 289 workers = [] 290 for i in range(self.num_workers): 291 slice1count = slice0count 292 if (i+1 == self.num_workers): 293 slice1count = left 294 ppslice = dict(itertools.islice(pp_set.items(), start, start+slice1count)) 295 p0 = mp.Process(target=work_slice, args=(i,self.batch_size,ppslice,self.gnunet_prefix)) 296 p0.start() 297 workers.append(p0) 298 start += slice1count 299 for w in workers: 300 w.join() 301 tend = time.time() 302 303 self.logger.info("Added %d RRSets for a total of %d RRs", pp_setcount, rrcount) 304 self.logger.info("All records have been added in %s seconds", 305 str(tend - tstart)) 306 307 308 def add_pkey_record_to_zone(self, pkey: str, domain: str, label: str, ttl: int) -> None: 309 """ 310 Adds the pkey of the subzone to the parent zone 311 :param pkey: the public key of the child zone 312 :param domain: the name of the parent zone 313 :param label: the label under which to add the pkey 314 :param ttl: the time to live the record should have in seconds 315 """ 316 data = ascension.util.classes.GNSRecordData( 317 value=pkey, 318 record_type='EDKEY', 319 relative_expiration=ttl, 320 is_relative_expiration=True, 321 is_private=not self.gnszone.public 322 ) 323 324 record_data = ascension.util.classes.GNSRRecordSet( 325 record_name=label, 326 data=[data] 327 ) 328 self.logger.debug("Added records to /namestore/%s with data %s", domain, record_data) 329 payload = record_data 330 self.ns_process.stdin.write(payload.record_name + "." + domain + ":\n") 331 for r in payload.data: 332 flags = "[r{}]".format('p' if not r.is_private else '') 333 # FIXME we have many more flags. but probably not in our use 334 # case? We always have relative expirations, for example. 335 self.ns_process.stdin.write("{} {} {} {}\n".format(r.record_type, 336 r.relative_expiration, 337 flags, 338 r.value)) 339 #FIXME error checking 340 #response = self.session.post(f"/namestore/{domain}", data=payload.to_json()) 341 342 #if response.status_code == 204: 343 # self.logger.debug("Added PKEY Record(s) with label %s", label) 344 # return 345 346 #resp = response.json() 347 #error = resp.get('error') 348 #self.logger.error("Task failed with error %s %s", 349 # error, 350 # ascension.util.rest.NAMESTORE_REST_API_ERRORS.get(error)) 351 352 353 def create_zone_hierarchy(self) -> None: 354 """ 355 Create equivalent DNS zone in GNS 356 This transformation is necessary as DNS zones are not equivalent to GNS zones 357 """ 358 # Extend Dictionary using GNS identities that already exist, 359 # checking for conflicts with information in DNS 360 self.logger.debug("Requesting all zones from the identity service") 361 response = self.session.get("/identity") 362 363 relevant_domains = list(filter( 364 lambda x: x['name'].endswith(self.gnszone.domain), 365 response.json()) 366 ) 367 for zone in relevant_domains: 368 self.subzonedict[zone['name']] = (zone['pubkey'], self.gnszone.minimum) 369 370 # Check if a delegated zone is available in GNS as per NS record 371 # Adds NS records that contain "gns--pkey--" to dictionary 372 nsrecords = self.dnszone.zone.iterate_rdatasets(dns.rdatatype.NS) 373 nameserverlist = [] 374 for nsrecord in nsrecords: 375 name = str(nsrecord[0]) 376 values = nsrecord[1] 377 ttl = values.ttl 378 379 # save DNS name object of nameservers for later 380 for nameserver in values: 381 nameserverlist.append(nameserver.target) 382 383 # filter for gns--pkey record in rdatas 384 gnspkeys = list(filter(lambda record: 385 str(record).startswith('gns--pkey--'), 386 values)) 387 num_gnspkeys = len(gnspkeys) 388 if not num_gnspkeys: 389 # skip empty values 390 continue 391 if num_gnspkeys > 1: 392 self.logger.critical( 393 "Detected ambiguous EDKEY records for label %s (not generating EDKEY record)", 394 name 395 ) 396 continue 397 gnspkey = str(gnspkeys[0]) 398 399 zonepkey = gnspkey[11:] 400 if len(zonepkey) != 59: 401 continue 402 403 zone = f"{name}.{self.gnszone.domain}" 404 if not self.subzonedict.get(zone): 405 self.subzonedict[zone] = (zonepkey, ttl) 406 else: 407 # This should be impossible!!? 408 pkey_ttl = self.subzonedict[zone] 409 pkey2, ttl = pkey_ttl 410 if pkey2 != gnspkey: 411 self.logger.critical("EDKEY in DNS does not match EDKEY in GNS for name %s", name) 412 continue 413 414 # Create missing zones (and add to dict) for GNS zones that are NOT DNS 415 # zones ("." in a label is not a zone-cut in DNS, but always in GNS). 416 # Only add the records for which there are no NS records for 417 remaining_nsrecords = set(filter(lambda name: not name.is_absolute(), 418 nameserverlist)) 419 remaining = set(filter(lambda name: name not in remaining_nsrecords, 420 self.dnszone.zone.nodes.keys())) 421 final = set(filter(lambda name: len(str(name).split('.')) > 1, 422 remaining)) 423 424 for name in final: 425 subzones = str(name).split('.') 426 for i in range(1, len(subzones)): 427 subdomain = ".".join(subzones[i:]) 428 zonename = f"{subdomain}.{self.gnszone.domain}" 429 ttl = self.gnszone.minimum # new record, cannot use existing one 430 if self.subzonedict.get(zonename) is None: 431 test = str(name) 432 if any(proto in test for proto in ('_tcp', '_udp')): 433 while not test.endswith(("_tcp", "_udp")): 434 chunk = test.split('.') 435 test = '.'.join(chunk[:-1]) 436 zonename = f"{chunk[-1]}.{self.gnszone.domain}" 437 self.subzonedict[zonename] = (None, ttl) 438 self.subzonedict[zonename] = (None, ttl) 439 continue 440 pkey = self.gnszone.create_zone_and_get_pkey(zonename) 441 self.subzonedict[zonename] = (pkey, ttl) 442 443 self.ns_process = subprocess.Popen(["gnunet-namestore", "-a", "-S"], stdin=subprocess.PIPE, text=True) 444 # Generate EDKEY records for all entries in subzonedict 445 for zone, pkeyttltuple in self.subzonedict.items(): 446 pkey, ttl = pkeyttltuple 447 # Allow for any amount of subzones 448 sub = zone.rstrip(self.gnszone.domain) 449 domain = ".".join(zone.split('.')[1:]) 450 # This happens if root is reached - can happen multiple times 451 if sub == '' or not pkeyttltuple[0]: 452 continue 453 label = zone.split('.')[0] 454 self.logger.info("Adding zone %s with %s zkey into %s", zone, pkey, domain) 455 self.add_pkey_record_to_zone(pkey, domain, label, int(ttl)) 456 self.ns_process.stdin.close() 457 self.ns_process.wait() 458 459 def purge_subzones(self): 460 for zname, zvalue in self.subzonedict: 461 self.gnszone.delete_zone(zname) 462 463 def main(): 464 """ 465 Initializes the Ascension class, handles arguments and daemon 466 """ 467 args = ascension.util.argumentparser.parse_arguments() 468 469 # Initialize class instance 470 ascender = Ascension(args) 471 472 # Attempt a zone transfer with the given arguments and keys 473 if args.dryrun: 474 transferrable = ascender.dnszone.test_zone_transfer() 475 if transferrable is None: 476 ascender.logger.critical( 477 'The specified domain is not transferrable using the given options!' 478 ) 479 return 1 480 ascender.logger.critical('SUCCESS! The specified domain is transferrable!') 481 return 0 482 483 # Checks if GNUnet REST API is running 484 #if not ascender.session.is_running(): 485 # ascender.logger.critical('GNUnet REST API is not reachable!') 486 487 # Set defaults to use before we get a SOA for the first time 488 retry = 300 489 490 # variable to keep state 491 needsupdate = False 492 first_run = True 493 494 # Main loop for actual daemon 495 while True: 496 gns_zone_serial = ascender.gnszone.get_gns_zone_serial() 497 498 ascender.logger.info("GNS zone serial is %s", gns_zone_serial) 499 dns_zone_serial = ascender.dnszone.get_dns_zone_serial() 500 ascender.logger.info("DNS zone serial is %s", dns_zone_serial) 501 502 if not dns_zone_serial: 503 ascender.logger.error("Could not get DNS zone serial") 504 if args.standalone: 505 return 1 506 time.sleep(retry) 507 continue 508 if not gns_zone_serial: 509 print("GNS zone does not exist yet, performing full transfer.") 510 ascender.gnszone.bootstrap_zone() 511 elif gns_zone_serial == dns_zone_serial: 512 print("GNS zone is up to date.") 513 if args.standalone: 514 return 0 515 time.sleep(retry) 516 elif gns_zone_serial > dns_zone_serial: 517 ascender.logger.critical("SOA serial in GNS is bigger than SOA serial in DNS?") 518 ascender.logger.critical("GNS zone: %s, DNS zone: %s", gns_zone_serial, dns_zone_serial) 519 if args.standalone: 520 return 1 521 time.sleep(retry) 522 continue 523 else: 524 print("GNS zone is out of date, performing incremental transfer.") 525 needsupdate = True 526 527 try: 528 start = time.time() 529 if not ascender.dnszone.zone or needsupdate: 530 # Zonebackups are needed for retaining information for IXFR and 531 # offer a zone for dnspython to patch 532 # On a first run, we may have been given an initial zone file to 533 # use as import source. 534 zf = None 535 if first_run: 536 zf = args.zonefile # May also be None 537 gns_zone_serial = ascender.dnszone.restore_from_file(gns_zone_serial, zonefile=zf) 538 ascender.logger.info("Zone serial for DNS zone transfer used: `%s'", gns_zone_serial) 539 # Transfer the actual zone 540 if None == zf: 541 ascender.dnszone.transfer_zone(gns_zone_serial) 542 ascender.dnszone.backup_to_file() 543 needsupdate = False 544 soa = ascender.dnszone.get_zone_soa() 545 end = time.time() 546 ascender.logger.info("Transferring the zone took %s seconds", str(end - start)) 547 retry = int(str(soa[2]).split(" ")[4]) 548 except dns.zone.BadZone: 549 ascender.logger.critical("Malformed DNS Zone '%s'", ascender.dnszone.domain) 550 if args.standalone: 551 return 2 552 time.sleep(retry) 553 continue 554 555 # FIXME: IXFR would require a "merge" 556 # FIXME: AXFR would require a GNS zone "purge" 557 # For now, even if we IXFR update the zone, perform a full update 558 ascender.gnszone.purge_records() 559 ascender.add_records_to_gns() 560 561 first_run = False 562 if args.standalone: 563 return 0 564 565 if __name__ == '__main__': 566 mp.set_start_method('spawn') 567 main()