#!/usr/bin/env python3 """Hosti DDNS 3.0: user-level installer and updater. Python 3.10+, standard library only.""" from __future__ import annotations import argparse, base64, configparser, ctypes, getpass, hashlib, ipaddress, json, logging from logging.handlers import RotatingFileHandler import os, pathlib, shutil, ssl, subprocess, sys, tempfile, time, urllib.error, urllib.parse, urllib.request VERSION='4.0.0' DEFAULT_ENDPOINT='https://ddns.hosti.me/DNS-Endpoint/update' DEFAULT_HOME=pathlib.Path.home()/'HostiDDNS' IS_WINDOWS=os.name=='nt' class ClientError(Exception): pass class NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self,req,fp,code,msg,headers,newurl): raise ClientError('Endpoint redirected. Set the final HTTPS endpoint in settings; the token was not forwarded.') def protect_key(value: str, decrypt=False) -> str: if not IS_WINDOWS: if decrypt and value.startswith('dpapi:'): raise ClientError('This token is protected for its original Windows user. Re-enter the token on this device.') return value class Blob(ctypes.Structure): _fields_=[('cbData',ctypes.c_ulong),('pbData',ctypes.POINTER(ctypes.c_ubyte))] try:data=base64.b64decode(value[6:],validate=True) if decrypt else value.encode('utf-8') except Exception as ex: raise ClientError('The saved Windows token is invalid. Re-enter it with --configure.') from ex buf=ctypes.create_string_buffer(data);incoming=Blob(len(data),ctypes.cast(buf,ctypes.POINTER(ctypes.c_ubyte)));out=Blob() function=ctypes.windll.crypt32.CryptUnprotectData if decrypt else ctypes.windll.crypt32.CryptProtectData function.argtypes=[ctypes.POINTER(Blob),ctypes.c_void_p,ctypes.c_void_p,ctypes.c_void_p,ctypes.c_void_p,ctypes.c_ulong,ctypes.POINTER(Blob)] function.restype=ctypes.c_int if not function(ctypes.byref(incoming),None,None,None,None,1,ctypes.byref(out)):raise ClientError('Windows token protection failed. Run as the same Windows user that installed the updater.') try: result=ctypes.string_at(out.pbData,out.cbData) return result.decode('utf-8') if decrypt else 'dpapi:'+base64.b64encode(result).decode('ascii') finally: ctypes.windll.kernel32.LocalFree.argtypes=[ctypes.c_void_p] ctypes.windll.kernel32.LocalFree.restype=ctypes.c_void_p ctypes.windll.kernel32.LocalFree(out.pbData) def read_settings(path: pathlib.Path) -> dict: parser=configparser.ConfigParser(interpolation=None) if not parser.read(path,encoding='utf-8'):raise ClientError('settings.conf was not found. Run the installer or configure the client first.') try: endpoint=parser['connection']['endpoint'].strip();hostname=parser['connection']['hostname'].strip().lower();key=parser['connection']['api_key'].strip() interval=parser.getint('updater','interval_seconds',fallback=300);address=parser.get('updater','address',fallback='auto').strip() except (KeyError,ValueError,configparser.Error) as ex:raise ClientError('settings.conf is incomplete or invalid.') from ex validate(endpoint,hostname,key,interval,address) if key.startswith('dpapi:'):key=protect_key(key,True) if not key or '\r' in key or '\n' in key:raise ClientError('Invalid API key.') return dict(endpoint=endpoint,hostname=hostname,api_key=key,interval_seconds=interval,address=address) def validate(endpoint,hostname,key,interval,address): u=urllib.parse.urlsplit(endpoint) if u.scheme!='https' or not u.hostname or u.username or u.password or u.fragment or u.query:raise ClientError('Endpoint must be an HTTPS URL without credentials, query or fragment.') if u.path.lower().rstrip('/') not in ['/dns-endpoint','/dns-endpoint/update']:raise ClientError('Endpoint must end with /DNS-Endpoint/ or /DNS-Endpoint/update.') if not hostname or len(hostname)>253 or any(not label or len(label)>63 or label[0]=='-' or label[-1]=='-' or any(c not in 'abcdefghijklmnopqrstuvwxyz0123456789-' for c in label) for label in hostname.split('.')) or '.' not in hostname:raise ClientError('Enter a complete hostname such as home.ddns.hosti.me.') if not key or '\r' in key or '\n' in key:raise ClientError('An API key / hostname token is required.') if not 60<=int(interval)<=86400:raise ClientError('Update interval must be 60–86400 seconds.') if address.lower()!='auto': try:ip=ipaddress.ip_address(address) except ValueError as ex:raise ClientError('Address must be auto or a public literal IP.') from ex if not ip.is_global:raise ClientError('Use a public IP address.') def write_settings(path: pathlib.Path,settings: dict): validate(settings['endpoint'],settings['hostname'],settings['api_key'],settings['interval_seconds'],settings['address']) parser=configparser.ConfigParser(interpolation=None) key=settings['api_key'];key=protect_key(key) if IS_WINDOWS and not key.startswith('dpapi:') else key parser['connection']={'endpoint':settings['endpoint'],'hostname':settings['hostname'],'api_key':key} parser['updater']={'interval_seconds':str(settings['interval_seconds']),'address':settings['address']} path.parent.mkdir(parents=True,exist_ok=True) fd,tmp=tempfile.mkstemp(prefix='.settings-',dir=path.parent) try: with os.fdopen(fd,'w',encoding='utf-8') as stream:parser.write(stream) os.chmod(tmp,0o600);os.replace(tmp,path) finally: if os.path.exists(tmp):os.unlink(tmp) def update(settings: dict) -> str: fields={'hostname':settings['hostname']} if settings['address'].lower()!='auto':fields['ip']=settings['address'] request=urllib.request.Request(settings['endpoint'],data=urllib.parse.urlencode(fields).encode('ascii'),headers={'Authorization':'Bearer '+settings['api_key'],'Content-Type':'application/x-www-form-urlencoded','User-Agent':'HostiDDNS/'+VERSION},method='POST') # Direct connections avoid accidentally updating to a system HTTP proxy's address. opener=urllib.request.build_opener(urllib.request.ProxyHandler({}),urllib.request.HTTPSHandler(context=ssl.create_default_context()),NoRedirect()) try: with opener.open(request,timeout=25) as response:body=response.read(4096).decode('utf-8','replace').strip() except urllib.error.HTTPError as ex: if ex.code==401:raise ClientError('API key rejected. Check token rotation, hostname and account access.') from None if ex.code==429:raise ClientError('Rate limit reached. Wait before retrying.') from None raise ClientError('Server returned HTTP '+str(ex.code)+'. Check the service and try later.') from None except (urllib.error.URLError,TimeoutError,ssl.SSLError,OSError):raise ClientError('Unable to reach the HTTPS endpoint. Check connection, certificate and service availability.') from None parts=body.split() if len(parts)!=2 or parts[0] not in ['good','nochg']:raise ClientError('Unexpected endpoint response. No successful update was recorded.') try:ipaddress.ip_address(parts[1]) except ValueError:raise ClientError('Endpoint returned an invalid IP address.') from None return ('Updated to ' if parts[0]=='good' else 'Address unchanged: ')+parts[1] def startup_name(directory):return 'HostiDDNS-'+hashlib.sha256(str(directory.resolve()).lower().encode()).hexdigest()[:12] def runner(directory): runtime=directory/'runtime'/'pythonw.exe' if runtime.exists():return [str(runtime),str(directory/'hosti_ddns.py'),'--daemon','--config',str(directory/'settings.conf')] executable=pathlib.Path(sys.executable) if IS_WINDOWS and executable.name.lower()=='python.exe' and executable.with_name('pythonw.exe').exists():executable=executable.with_name('pythonw.exe') return [str(executable),str(directory/'hosti_ddns.py'),'--daemon','--config',str(directory/'settings.conf')] def startup(directory,enable): if not IS_WINDOWS:raise ClientError('Automatic sign-in startup is supported on Windows. Use your OS scheduler elsewhere.') import winreg with winreg.CreateKey(winreg.HKEY_CURRENT_USER,r'Software\Microsoft\Windows\CurrentVersion\Run') as key: name=startup_name(directory) if enable:winreg.SetValueEx(key,name,0,winreg.REG_SZ,subprocess.list2cmdline(runner(directory))) else: try:winreg.DeleteValue(key,name) except FileNotFoundError:pass def yes(prompt,default=True): answer=input(prompt+(' [Y/n]: ' if default else ' [y/N]: ')).strip().lower() return answer in ['y','yes'] or not answer and default def wizard_values(existing=None): old=existing or {} endpoint=input('HTTPS endpoint ['+old.get('endpoint',DEFAULT_ENDPOINT)+']: ').strip() or old.get('endpoint',DEFAULT_ENDPOINT) hostname=input('Hostname'+(' ['+old['hostname']+']' if old.get('hostname') else '')+': ').strip().lower() or old.get('hostname','') key=getpass.getpass('API key / hostname token'+(' [Enter to keep current]' if old.get('api_key') else '')+': ').strip() or old.get('api_key','') interval=int(input('Update interval in seconds ['+str(old.get('interval_seconds',300))+']: ').strip() or old.get('interval_seconds',300)) address=input('Public IP, or auto ['+old.get('address','auto')+']: ').strip() or old.get('address','auto') validate(endpoint,hostname,key,interval,address) print('This token will be sent only to: '+urllib.parse.urlsplit(endpoint).hostname) if not yes('Save these settings?'):raise ClientError('Settings were not saved.') return dict(endpoint=endpoint,hostname=hostname,api_key=key,interval_seconds=interval,address=address) def restrict_windows(path): if not IS_WINDOWS:return # SID lookup avoids username / domain ambiguity, and works on local and AD accounts. import csv,io row=next(csv.reader(io.StringIO(subprocess.check_output(['whoami','/user','/fo','csv','/nh'],text=True)))) sid=row[-1] result=subprocess.run(['icacls',str(path),'/inheritance:r','/grant:r','*'+sid+':(OI)(CI)F','*S-1-5-18:(OI)(CI)F','*S-1-5-32-544:(OI)(CI)F'],capture_output=True,text=True) if result.returncode:raise ClientError('Could not restrict the install directory permissions. Choose an NTFS folder you own.') def install(directory: pathlib.Path,import_file=None,offer_background=True): directory=directory.expanduser().resolve() if directory==pathlib.Path.home().resolve() or directory==directory.parent:raise ClientError('Choose an application subfolder, not your home folder or drive root.') marker=directory/'hosti-install.json' if directory.exists() and any(directory.iterdir()) and not marker.exists():raise ClientError('Choose an empty custom folder. Existing unrelated files will not be overwritten.') if marker.exists(): if not yes('An installation already exists here. Update its client files?',False):return directory.mkdir(parents=True,exist_ok=True);restrict_windows(directory) source=pathlib.Path(__file__).resolve();bundled=source.parent/'runtime' # Stop the existing daemon before replacing any of its client files. (directory/'stop.flag').write_text('stop',encoding='ascii') if (directory/'updater.lock').exists(): for _ in range(30): if not (directory/'updater.lock').exists():break time.sleep(.2) if (directory/'updater.lock').exists():raise ClientError('The updater is still running. Close it before installing new client files.') if bundled.exists() and bundled.resolve()!=(directory/'runtime').resolve():shutil.copytree(bundled,directory/'runtime',dirs_exist_ok=True) if source!=(directory/'hosti_ddns.py').resolve():shutil.copy2(source,directory/'hosti_ddns.py') exe=os.environ.get('HOSTI_INSTALLER_EXE','') if exe and pathlib.Path(exe).is_file() and pathlib.Path(exe).resolve()!=(directory/'HostiDDNS.exe').resolve():shutil.copy2(exe,directory/'HostiDDNS.exe') marker.write_text(json.dumps({'product':'HostiDDNS','version':VERSION}),encoding='utf-8') settings_file=directory/'settings.conf';settings=None if import_file:settings=read_settings(pathlib.Path(import_file));print('Imported settings for '+settings['hostname']) elif settings_file.exists():settings=read_settings(settings_file) settings=wizard_values(settings) write_settings(settings_file,settings) print('Installed in: '+str(directory));print('Configuration: '+str(settings_file)) if IS_WINDOWS: automatic=yes('Start automatically when this Windows user signs in?') startup(directory,automatic) print('Automatic startup '+('enabled.' if automatic else 'disabled.')) (directory/'stop.flag').unlink(missing_ok=True) if yes('Test an update now?'): try:print(update(settings)) except ClientError as ex:print('Test failed: '+str(ex)+' Settings are saved; configure them again if needed.') if offer_background and yes('Start the background updater now?'): kwargs={'creationflags':subprocess.CREATE_NO_WINDOW} if IS_WINDOWS else {'start_new_session':True} subprocess.Popen(runner(directory),stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL,**kwargs) print('Background updater started. See updater.log in your install folder.') def daemon(settings_file): directory=settings_file.parent;lock=directory/'updater.lock' # Exclusive creation gives one updater per install. PID used to recover stale locks on Windows. try:fd=os.open(lock,os.O_CREAT|os.O_EXCL|os.O_WRONLY,0o600) except FileExistsError: try: pid=int(lock.read_text()) if IS_WINDOWS: ctypes.windll.kernel32.OpenProcess.restype=ctypes.c_void_p handle=ctypes.windll.kernel32.OpenProcess(0x1000,False,pid) if handle:ctypes.windll.kernel32.CloseHandle(ctypes.c_void_p(handle));return else: os.kill(pid,0);return except (ValueError,OSError):pass lock.unlink(missing_ok=True) try:fd=os.open(lock,os.O_CREAT|os.O_EXCL|os.O_WRONLY,0o600) except FileExistsError:return os.write(fd,str(os.getpid()).encode());os.close(fd) logger=logging.getLogger('hosti');logger.setLevel(logging.INFO) handler=RotatingFileHandler(directory/'updater.log',maxBytes=256000,backupCount=2,encoding='utf-8');handler.setFormatter(logging.Formatter('%(asctime)s %(message)s'));logger.addHandler(handler) logger.info('Hosti DDNS '+VERSION+' started. Tokens are not logged.') try: while not (directory/'stop.flag').exists(): delay=300 try: settings=read_settings(settings_file);delay=settings['interval_seconds'];logger.info(update(settings)) except (ClientError,configparser.Error,ValueError,OSError) as ex: # Do not log raw parsing / OS exceptions that might include configuration values. logger.error(str(ex) if isinstance(ex,ClientError) else 'Settings or local file error. Run configure to repair it.') delay=max(delay,300) deadline=time.monotonic()+delay while time.monotonic()