#!/usr/bin/python3 # -*- coding: utf-8 -*- """ Configurator for Home Assistant. https://github.com/danielperna84/hass-poc-configurator """ import os import sys import json import ssl import socketserver import base64 import ipaddress import signal import urllib.request from string import Template from http.server import BaseHTTPRequestHandler, HTTPServer from urllib.parse import urlparse, parse_qs, unquote ### Some options for you to change LISTENIP = "0.0.0.0" LISTENPORT = 3218 # Set BASEPATH to something like "/home/hass/.homeasssitant" if you're not running the configurator from that path BASEPATH = None # Set the paths to a certificate and the key if you're using SSL, e.g "/etc/ssl/certs/mycert.pem" SSL_CERTIFICATE = None SSL_KEY = None # Set the destination where the HASS API is reachable HASS_API = "http://127.0.0.1:8123/api/" # If a password is required to access the API, set it in the form of "password" HASS_API_PASSWORD = None # To enable authentication, set the credentials in the form of "username:password" CREDENTIALS = None # Limit access to the configurator by adding allowed IP addresses / networks to the list, # e.g ALLOWED_NETWORKS = ["192.168.0.0/24", "172.16.47.23"] ALLOWED_NETWORKS = [] # List of statically banned IP addresses, e.g. ["1.1.1.1", "2.2.2.2"] BANNED_IPS = [] # Ban IPs after n failed login attempts. Restart service to reset banning. The default of `0` disables this feature. BANLIMIT = 0 ### End of options RELEASEURL = "https://api.github.com/repos/danielperna84/hass-poc-configurator/releases/latest" VERSION = "0.1.2" BASEDIR = "." DEV = False HTTPD = None FAIL2BAN_IPS = {} INDEX = Template(r""" HASS Configurator


    • Editor Settings

      • Keyboard Shortcuts

      Save Settings Locally

      Ace Editor 1.2.6

    """) def signal_handler(signal, frame): global HTTPD print("Shutting down server") HTTPD.server_close() sys.exit(0) def load_settings(settingsfile): global LISTENIP, LISTENPORT, BASEPATH, SSL_CERTIFICATE, SSL_KEY, HASS_API, \ HASS_API_PASSWORD, CREDENTIALS, ALLOWED_NETWORKS, BANNED_IPS, BANLIMIT try: if os.path.isfile(settingsfile): with open(settingsfile) as fptr: settings = json.loads(fptr.read()) LISTENIP = settings.get("LISTENIP", LISTENIP) LISTENPORT = settings.get("LISTENPORT", LISTENPORT) BASEPATH = settings.get("BASEPATH", BASEPATH) SSL_CERTIFICATE = settings.get("SSL_CERTIFICATE", SSL_CERTIFICATE) SSL_KEY = settings.get("SSL_KEY", SSL_KEY) HASS_API = settings.get("HASS_API", HASS_API) HASS_API_PASSWORD = settings.get("HASS_API_PASSWORD", HASS_API_PASSWORD) CREDENTIALS = settings.get("CREDENTIALS", CREDENTIALS) ALLOWED_NETWORKS = settings.get("ALLOWED_NETWORKS", ALLOWED_NETWORKS) BANNED_IPS = settings.get("BANNED_IPS", BANNED_IPS) BANLIMIT = settings.get("BANLIMIT", BANLIMIT) except Exception as err: print(err) print("Not loading static settings") return False def get_dircontent(path): dircontent = [] for e in sorted(os.listdir(path), key=lambda x: x.lower()): edata = {} edata['name'] = e edata['dir'] = path edata['fullpath'] = os.path.abspath(os.path.join(path, e)) edata['type'] = 'dir' if os.path.isdir(edata['fullpath']) else 'file' try: stats = os.stat(os.path.join(path, e)) edata['size'] = stats.st_size edata['modified'] = stats.st_mtime except Exception: edata['size'] = 0 edata['modified'] = 0 dircontent.append(edata) return dircontent def get_html(): if DEV: try: with open("dev.html") as fptr: html = Template(fptr.read()) return html except Exception as err: print(err) print("Delivering embedded HTML") return INDEX def check_access(clientip): global BANNED_IPS if clientip in BANNED_IPS: return False if not ALLOWED_NETWORKS: return True for net in ALLOWED_NETWORKS: ipobject = ipaddress.ip_address(clientip) if ipobject in ipaddress.ip_network(net): return True BANNED_IPS.append(clientip) return False class RequestHandler(BaseHTTPRequestHandler): def do_BLOCK(self): self.send_response(420) self.end_headers() self.wfile.write(bytes("Policy not fulfilled", "utf8")) def do_GET(self): if not check_access(self.client_address[0]): self.do_BLOCK() return req = urlparse(self.path) query = parse_qs(req.query) self.send_response(200) if req.path == '/api/file': content = "" self.send_header('Content-type','text/text') self.end_headers() filename = query.get('filename', None) try: if filename: filename = unquote(filename[0]).encode('utf-8') print(filename) if os.path.isfile(os.path.join(BASEDIR.encode('utf-8'), filename)): with open(os.path.join(BASEDIR.encode('utf-8'), filename)) as fptr: content += fptr.read() else: content = "File not found" except Exception as err: print(err) content = str(err) self.wfile.write(bytes(content, "utf8")) return elif req.path == '/api/listdir': content = "" self.send_header('Content-type','text/json') self.end_headers() dirpath = query.get('path', None) try: if dirpath: dirpath = unquote(dirpath[0]).encode('utf-8') if os.path.isdir(dirpath): dircontent = get_dircontent(dirpath.decode('utf-8')) filedata = {'content': dircontent, 'abspath': os.path.abspath(dirpath).decode('utf-8'), 'parent': os.path.dirname(os.path.abspath(dirpath)).decode('utf-8') } self.wfile.write(bytes(json.dumps(filedata), "utf8")) except Exception as err: print(err) content = str(err) self.wfile.write(bytes(content, "utf8")) return elif req.path == '/api/abspath': content = "" self.send_header('Content-type','text/text') self.end_headers() dirpath = query.get('path', None) if dirpath: dirpath = unquote(dirpath[0]).encode('utf-8') print(dirpath) absp = os.path.abspath(dirpath) print(absp) if os.path.isdir(dirpath): self.wfile.write(os.path.abspath(dirpath)) return elif req.path == '/api/parent': content = "" self.send_header('Content-type','text/text') self.end_headers() dirpath = query.get('path', None) if dirpath: dirpath = unquote(dirpath[0]).encode('utf-8') print(dirpath) absp = os.path.abspath(dirpath) print(absp) if os.path.isdir(dirpath): self.wfile.write(os.path.abspath(os.path.dirname(dirpath))) return elif req.path == '/api/restart': print("/api/restart") self.send_header('Content-type','text/json') self.end_headers() r = {"restart": False} try: headers = { "Content-Type": "application/json" } if HASS_API_PASSWORD: headers["x-ha-access"] = HASS_API_PASSWORD req = urllib.request.Request("%sservices/homeassistant/restart" % HASS_API, headers=headers, method='POST') with urllib.request.urlopen(req) as response: r = json.loads(response.read().decode('utf-8')) print(r) except Exception as err: print(err) r['restart'] = str(err) self.wfile.write(bytes(json.dumps(r), "utf8")) return elif req.path == '/': self.send_header('Content-type','text/html') self.end_headers() boot = "{}" try: headers = { "Content-Type": "application/json" } if HASS_API_PASSWORD: headers["x-ha-access"] = HASS_API_PASSWORD req = urllib.request.Request("%sbootstrap" % HASS_API, headers=headers, method='GET') with urllib.request.urlopen(req) as response: boot = response.read().decode('utf-8') except Exception as err: print("Exception getting bootstrap") print(err) color = "green" try: response = urllib.request.urlopen(RELEASEURL) latest = json.loads(response.read().decode('utf-8'))['tag_name'] if VERSION != latest: color = "red" except Exception as err: print("Exception getting release") print(err) html = get_html().safe_substitute(bootstrap=boot, current=VERSION, versionclass=color, separator="\%s" % os.sep if os.sep == "\\" else os.sep) self.wfile.write(bytes(html, "utf8")) return else: self.send_response(404) self.end_headers() self.wfile.write(bytes("File not found", "utf8")) def do_POST(self): if not check_access(self.client_address[0]): self.do_BLOCK() return req = urlparse(self.path) postvars = {} response = { "error": True, "message": "Generic failure" } try: length = int(self.headers['content-length']) postvars = parse_qs(self.rfile.read(length).decode('utf-8'), keep_blank_values=1) interror = False except Exception as err: print(err) response['message'] = "%s" % (str(err)) interror = True if not interror: if req.path == '/api/save': if 'filename' in postvars.keys() and 'text' in postvars.keys(): if postvars['filename'] and postvars['text']: try: filename = unquote(postvars['filename'][0]) response['file'] = filename with open(filename, 'wb') as fptr: fptr.write(bytes(postvars['text'][0], "utf-8")) self.send_response(200) self.send_header('Content-type','text/json') self.end_headers() response['error'] = False response['message'] = "File saved successfully" self.wfile.write(bytes(json.dumps(response), "utf8")) return except Exception as err: response['message'] = "%s" % (str(err)) print(err) else: response['message'] = "Missing filename or text" elif req.path == '/api/delete': if 'path' in postvars.keys(): if postvars['path']: try: delpath = unquote(postvars['path'][0]) response['path'] = delpath try: if os.path.isdir(delpath): os.rmdir(delpath) else: os.unlink(delpath) self.send_response(200) self.send_header('Content-type','text/json') self.end_headers() response['error'] = False response['message'] = "Deletetion successful" self.wfile.write(bytes(json.dumps(response), "utf8")) return except Exception as err: print(err) response['error'] = True response['message'] = str(err) except Exception as err: response['message'] = "%s" % (str(err)) print(err) else: response['message'] = "Missing filename or text" elif req.path == '/api/newfolder': if 'path' in postvars.keys() and 'name' in postvars.keys(): if postvars['path'] and postvars['name']: try: basepath = unquote(postvars['path'][0]) name = unquote(postvars['name'][0]) response['path'] = os.path.join(basepath, name) try: os.makedirs(response['path']) self.send_response(200) self.send_header('Content-type','text/json') self.end_headers() response['error'] = False response['message'] = "Folder created" self.wfile.write(bytes(json.dumps(response), "utf8")) return except Exception as err: print(err) response['error'] = True response['message'] = str(err) except Exception as err: response['message'] = "%s" % (str(err)) print(err) else: response['message'] = "Missing filename or text" else: response['message'] = "Invalid method" self.send_response(200) self.send_header('Content-type','text/json') self.end_headers() self.wfile.write(bytes(json.dumps(response), "utf8")) return class AuthHandler(RequestHandler): def do_AUTHHEAD(self): print("Requesting authorization") self.send_response(401) self.send_header('WWW-Authenticate', 'Basic realm=\"HASS-PoC-Configurator\"') self.send_header('Content-type', 'text/html') self.end_headers() def do_GET(self): global CREDENTIALS authorization = self.headers.get('Authorization', None) if authorization == None: self.do_AUTHHEAD() self.wfile.write(bytes('no auth header received', 'utf-8')) pass elif authorization == 'Basic %s' % CREDENTIALS.decode('utf-8'): if BANLIMIT: FAIL2BAN_IPS.pop(self.client_address[0], None) super().do_GET() pass else: if BANLIMIT: bancounter = FAIL2BAN_IPS.get(self.client_address[0], 1) if bancounter >= BANLIMIT: print("Blocking access from %s" % self.client_address[0]) self.do_BLOCK() return else: FAIL2BAN_IPS[self.client_address[0]] = bancounter + 1 self.do_AUTHHEAD() self.wfile.write(bytes('Authentication required', 'utf-8')) pass def do_POST(self): global CREDENTIALS authorization = self.headers.get('Authorization', None) if authorization == None: self.do_AUTHHEAD() self.wfile.write(bytes('no auth header received', 'utf-8')) pass elif authorization == 'Basic %s' % CREDENTIALS.decode('utf-8'): if BANLIMIT: FAIL2BAN_IPS.pop(self.client_address[0], None) super().do_POST() pass else: if BANLIMIT: bancounter = FAIL2BAN_IPS.get(self.client_address[0], 1) if bancounter >= BANLIMIT: print("Blocking access from %s" % self.client_address[0]) self.do_BLOCK() return else: FAIL2BAN_IPS[self.client_address[0]] = bancounter + 1 self.do_AUTHHEAD() self.wfile.write(bytes('Authentication required', 'utf-8')) pass def main(args): global HTTPD, CREDENTIALS if args: load_settings(args[0]) print("Starting server") server_address = (LISTENIP, LISTENPORT) if CREDENTIALS: CREDENTIALS = base64.b64encode(bytes(CREDENTIALS, "utf-8")) Handler = AuthHandler else: Handler = RequestHandler if not SSL_CERTIFICATE: HTTPD = HTTPServer(server_address, Handler) else: HTTPD = socketserver.TCPServer(server_address, Handler) HTTPD.socket = ssl.wrap_socket(HTTPD.socket, certfile=SSL_CERTIFICATE, keyfile=SSL_KEY, server_side=True) print('Listening on: %s://%s:%i' % ('https' if SSL_CERTIFICATE else 'http', LISTENIP, LISTENPORT)) if BASEPATH: os.chdir(BASEPATH) HTTPD.serve_forever() if __name__ == "__main__": signal.signal(signal.SIGINT, signal_handler) main(sys.argv[1:])