Created
July 30, 2025 08:19
-
-
Save pgdad/efc1e4ebef2a0313e91fe4fcbd0f800c to your computer and use it in GitHub Desktop.
Python utilities for fetching files from DNS TXT records
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| #!/usr/bin/env python3 | |
| """ | |
| fetch_file.py - Fetch a single file from DNS TXT records | |
| Usage: python3 fetch_file.py <filename> [output_directory] [index_hostname] | |
| Environment: PORKBUN_INDEX (default: index) | |
| """ | |
| import argparse | |
| import json | |
| import os | |
| import subprocess | |
| import sys | |
| def lookup_txt_record(domain): | |
| """Lookup TXT record using nslookup""" | |
| try: | |
| result = subprocess.run( | |
| ['nslookup', '-type=TXT', domain], | |
| capture_output=True, | |
| text=True, | |
| timeout=10 | |
| ) | |
| if result.returncode != 0: | |
| return None | |
| # Find line with 'text = ' | |
| for line in result.stdout.splitlines(): | |
| if 'text = ' in line: | |
| # Extract content between quotes and unescape | |
| content = line.split('text = "', 1)[1].rsplit('"', 1)[0] | |
| content = content.replace('\\"', '"') | |
| return content | |
| return None | |
| except (subprocess.TimeoutExpired, subprocess.SubprocessError, FileNotFoundError): | |
| return None | |
| def unescape_content(content): | |
| """Unescape octal sequences in content""" | |
| # Replace octal escape sequences | |
| content = content.replace('\\010', '\n') # newline | |
| content = content.replace('\\009', '\t') # tab | |
| return content | |
| def main(): | |
| parser = argparse.ArgumentParser( | |
| description="Fetch a single file from DNS TXT records", | |
| formatter_class=argparse.RawDescriptionHelpFormatter, | |
| epilog=""" | |
| Examples: | |
| python3 fetch_file.py myfile.txt | |
| python3 fetch_file.py myfile.txt /path/to/output | |
| python3 fetch_file.py myfile.txt /path/to/output myindex | |
| Environment variables: | |
| PORKBUN_INDEX - Default index hostname (default: index) | |
| """.strip() | |
| ) | |
| parser.add_argument('filename', help='Name of file to fetch') | |
| parser.add_argument('output_dir', nargs='?', default='porkbun-downloads', | |
| help='Output directory (default: porkbun-downloads)') | |
| parser.add_argument('index_host', nargs='?', | |
| help='Index hostname (default: from PORKBUN_INDEX env var or "index")') | |
| args = parser.parse_args() | |
| # Get index hostname | |
| index_host = args.index_host or os.environ.get('PORKBUN_INDEX', 'index') | |
| index_fqdn = f"{index_host}.pgdad.org" | |
| print(f"Reading index from DNS ({index_fqdn})...", file=sys.stderr) | |
| # Read index from DNS | |
| index_content = lookup_txt_record(index_fqdn) | |
| if not index_content: | |
| print("Error: Could not read index from DNS", file=sys.stderr) | |
| sys.exit(1) | |
| # Parse JSON | |
| try: | |
| index = json.loads(index_content) | |
| except json.JSONDecodeError: | |
| print("Error: Invalid JSON in index", file=sys.stderr) | |
| sys.exit(1) | |
| # Find file entry | |
| if args.filename not in index: | |
| print(f"Error: File '{args.filename}' not found in index", file=sys.stderr) | |
| sys.exit(1) | |
| entry = index[args.filename] | |
| suffix = entry['suffix'] | |
| # Extract subindex from suffix | |
| subindex = suffix.replace('content.pgdad.org', '') | |
| print(f"Found file '{args.filename}' with subindex: {subindex}", file=sys.stderr) | |
| # Fetch content chunks sequentially | |
| print("Fetching content chunks...", file=sys.stderr) | |
| record_index = 0 | |
| content = "" | |
| while True: | |
| record_name = f"c{record_index}-{subindex}content.pgdad.org" | |
| chunk = lookup_txt_record(record_name) | |
| if not chunk: | |
| break | |
| content += chunk | |
| record_index += 1 | |
| print(f" Fetched chunk {record_index}", file=sys.stderr) | |
| if record_index == 0: | |
| print(f"Error: No content records found for file '{args.filename}'", file=sys.stderr) | |
| sys.exit(1) | |
| print(f"Successfully fetched {record_index} chunks", file=sys.stderr) | |
| # Create output directory if it doesn't exist | |
| if not os.path.exists(args.output_dir): | |
| print(f"Creating output directory: {args.output_dir}", file=sys.stderr) | |
| os.makedirs(args.output_dir, exist_ok=True) | |
| # Write content to file (unescaping octal sequences) | |
| output_file = os.path.join(args.output_dir, args.filename) | |
| content = unescape_content(content) | |
| with open(output_file, 'w', encoding='utf-8') as f: | |
| f.write(content) | |
| print(f"File saved to: {output_file}", file=sys.stderr) | |
| if __name__ == "__main__": | |
| main() |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| #!/usr/bin/env python3 | |
| """ | |
| list_files.py - List all stored files from DNS TXT records | |
| Usage: python3 list_files.py [index_hostname] | |
| Environment: PORKBUN_INDEX (default: index) | |
| """ | |
| import argparse | |
| import json | |
| import os | |
| import socket | |
| import subprocess | |
| import sys | |
| def lookup_txt_record(domain): | |
| """Lookup TXT record using nslookup""" | |
| try: | |
| result = subprocess.run( | |
| ['nslookup', '-type=TXT', domain], | |
| capture_output=True, | |
| text=True, | |
| timeout=10 | |
| ) | |
| if result.returncode != 0: | |
| return None | |
| # Find line with 'text = ' | |
| for line in result.stdout.splitlines(): | |
| if 'text = ' in line: | |
| # Extract content between quotes and unescape | |
| content = line.split('text = "', 1)[1].rsplit('"', 1)[0] | |
| content = content.replace('\\"', '"') | |
| return content | |
| return None | |
| except (subprocess.TimeoutExpired, subprocess.SubprocessError, FileNotFoundError): | |
| return None | |
| def main(): | |
| parser = argparse.ArgumentParser( | |
| description="List all stored files from DNS TXT records", | |
| formatter_class=argparse.RawDescriptionHelpFormatter, | |
| epilog=""" | |
| Examples: | |
| python3 list_files.py | |
| python3 list_files.py myindex | |
| Environment variables: | |
| PORKBUN_INDEX - Default index hostname (default: index) | |
| """.strip() | |
| ) | |
| parser.add_argument('index_host', nargs='?', | |
| help='Index hostname (default: from PORKBUN_INDEX env var or "index")') | |
| args = parser.parse_args() | |
| # Get index hostname | |
| index_host = args.index_host or os.environ.get('PORKBUN_INDEX', 'index') | |
| index_fqdn = f"{index_host}.pgdad.org" | |
| print(f"Reading index from DNS ({index_fqdn})...", file=sys.stderr) | |
| # Read index from DNS | |
| index_content = lookup_txt_record(index_fqdn) | |
| if not index_content: | |
| print("Error: Could not read index from DNS", file=sys.stderr) | |
| sys.exit(1) | |
| # Parse JSON | |
| try: | |
| index = json.loads(index_content) | |
| except json.JSONDecodeError: | |
| print("Error: Invalid JSON in index", file=sys.stderr) | |
| sys.exit(1) | |
| # Check if index is empty | |
| if not index: | |
| print("No files stored.") | |
| return | |
| # Sort filenames and display | |
| print("Stored files:") | |
| for filename in sorted(index.keys()): | |
| entry = index[filename] | |
| suffix = entry['suffix'] | |
| encrypted = entry.get('encrypted', False) | |
| encrypted_str = " [encrypted]" if encrypted else "" | |
| print(f" {filename} -> {suffix}{encrypted_str}") | |
| if __name__ == "__main__": | |
| main() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment