diff --git a/api/config.py b/api/config.py new file mode 100644 index 0000000..a6713ac --- /dev/null +++ b/api/config.py @@ -0,0 +1,59 @@ +import os + +class ConfigManager: + def __init__(self, file_path: str): + self.file_path = file_path + if not os.path.exists(self.file_path): + with open(self.file_path, 'w') as f: + pass + + def __getattr__(self, key): + return self.get(key) + + def __setattr__(self, key, value): + if key in ['file_path']: + super().__setattr__(key, value) + else: + self.set(key, value) + + def __iter__(self): + return iter(self.to_dict().items()) + + def to_dict(self): + result = {} + with open(self.file_path, 'r') as f: + for line in f: + if line.startswith('#'): + continue + if '=' in line: + key, value = line.strip().split('=', 1) + result[key] = value + return result + + def get(self, key): + with open(self.file_path, 'r') as f: + for line in f: + if line.startswith('#'): + continue + if line.startswith(key + '='): + return line[len(key) + 1:].strip() + return None + + def set(self, key, value): + lines = [] + found = False + new_value = f'{key}={value}\n' + with open(self.file_path, 'r') as f: + for line in f: + if line.startswith(key + '=') and not line.startswith('#'): + if line == new_value: + return True + # update value instead of adding current line + lines.append(new_value) + found = True + else: + lines.append(line) + if not found: + lines.append(new_value) + with open(self.file_path, 'w') as f: + f.writelines(lines) diff --git a/api/const.py b/api/const.py new file mode 100644 index 0000000..1d342b1 --- /dev/null +++ b/api/const.py @@ -0,0 +1,5 @@ +services_dir = '/etc/init.d' +services_ignored = ['boot', 'coredump', 'done', 'led', 'silentboot'] +config_listener = '/data/listener' + +lx06_infrared = '/sys/ir_tx_gpio/ir_data' \ No newline at end of file diff --git a/api/main.py b/api/main.py index e67e1a1..105f6a8 100644 --- a/api/main.py +++ b/api/main.py @@ -1,29 +1,151 @@ -from flask import Flask, jsonify, request +from flask import Flask, jsonify, request, redirect import os import re +import requests +import subprocess + +from config import ConfigManager +from utils import get_ip_address +import const hostname = os.uname()[1] +speaker_ip = get_ip_address('wlan0') app = Flask(__name__) -services_dir = '/etc/init.d' +config = ConfigManager(const.config_listener) @app.route('/') +def index(): + return app.send_static_file('index.html') + +@app.get('/discover') def info(): - return jsonify({'hostname': hostname}) + service_search_name = '_home-assistant._tcp' + def parse_avahi_output(output): + instances = list() + for line in output.split('\n'): + if service_search_name in line and 'IPv4' in line: + parts = line.split(';') + if len(parts) > 7: + service_name = parts[3] + txt_records = parts[9] + txt_dict = {} + for txt in re.findall(r'"(.*?)"', txt_records): + if '=' in txt: + key, val = txt.split('=', 1) + if val == 'True': + val = True + if val == 'False': + val = False + txt_dict[key] = val + instances.append(txt_dict) + return instances + + try: + result = subprocess.run(f'avahi-browse -rpt {service_search_name}'.split(' '), capture_output=True, text=True) + if result.returncode == 0: + instances = parse_avahi_output(result.stdout) + return jsonify({'hostname': hostname, 'instances': instances}) + else: + return jsonify({'hostname': hostname, 'instances': []}), 500 + except Exception as e: + return jsonify({'hostname': hostname, 'error': str(e)}), 500 + +@app.post('/auth') +def home_assistant_auth(): + ha_url = request.form.get('url', '').rstrip('/') + if not ha_url or not ha_url.startswith('http'): + return jsonify({'error': 'Missing url parameter'}), 400 + + if config.HA_URL == ha_url and config.HA_TOKEN and len(config.HA_TOKEN) > 30: + return jsonify({'message': 'Instance already configured'}) + + config.HA_URL = ha_url + config.HA_TOKEN = "none" + + data = { + 'client_id': f'http://{speaker_ip}', + 'redirect_uri': f'http://{speaker_ip}/auth_callback', + } + + query_params = '&'.join([f'{key}={value}'.replace(':','%3A').replace('/','%2F') for key, value in data.items()]) + return redirect(f'{ha_url}/auth/authorize?{query_params}', code=303) + +# https://developers.home-assistant.io/docs/auth_api/ +@app.get('/auth_callback') +def home_assistant_auth_callback(): + code = request.args.get('code') + store_token = request.args.get('storeToken', 'false').lower() == 'true' + state = request.args.get('state') + + if not code: + return jsonify({'error': 'Missing code parameter'}), 400 + + data = { + 'grant_type': 'authorization_code', + 'code': code, + 'client_id': f'http://{speaker_ip}', + } + + headers = { + 'Content-Type': 'application/x-www-form-urlencoded' + } + + ha_url = config.HA_URL + if not ha_url: + return jsonify({'error': 'Home Assistant URL not configured'}), 500 + + req = requests.post(f'{ha_url}/auth/token', data=data, headers=headers) + + if req.status_code != 200: + return jsonify({'error': 'Failed to get access token', 'code': req.status_code, 'response': req.json()}), 500 + + token = req.json() + if store_token: + config.HA_TOKEN = token['access_token'] + config.HA_REFRESH_TOKEN = token['refresh_token'] + + return jsonify({'message': 'Auth configured'}) + +def home_assistant_refresh_token(): + ha_url = config.HA_URL + if not ha_url: + return False + + data = { + 'grant_type': 'refresh_token', + 'refresh_token': config.HA_REFRESH_TOKEN, + } + + headers = { + 'Content-Type': 'application/x-www-form-urlencoded' + } + + req = requests.post(f'{ha_url}/auth/token', data=data, headers=headers) + + if req.status_code != 200: + return False + + token = req.json() + config.HA_TOKEN = token['access_token'] + + return True @app.route('/services') def list_services(): - ignored_list = ['boot', 'coredump', 'done', 'led', 'silentboot'] - files = [f for f in os.listdir(services_dir) if os.access(os.path.join(services_dir, f), os.X_OK) and f not in ignored_list] + files = [f for f in os.listdir(const.services_dir) if os.access(os.path.join(const.services_dir, f), os.X_OK) and f not in const.services_ignored] return jsonify({'data': {'services': files}}) @app.route('/services//') def manage_service(service, action): action = action.lower().strip() - service_path = os.path.join(services_dir, service) + service_path = os.path.join(const.services_dir, service) if not os.path.exists(service_path) or not os.access(service_path, os.X_OK): return jsonify({'error': 'Service not found or not executable'}), 404 + if service in const.services_ignored: + return jsonify({'error': 'Service not allowed'}), 403 + if action not in ['start', 'stop', 'restart']: return jsonify({'error': 'Invalid action'}), 400 @@ -50,7 +172,7 @@ def send_infrared(): return jsonify({'error': 'Invalid code format'}), 400 try: - with open('/sys/ir_tx_gpio/ir_data', 'w') as f: + with open(const.lx06_infrared, 'w') as f: f.write(code) except IOError as e: return jsonify({'error': f'Failed to send infrared signal: {e}'}), 500 diff --git a/api/static/index.html b/api/static/index.html new file mode 100644 index 0000000..49eef50 --- /dev/null +++ b/api/static/index.html @@ -0,0 +1,55 @@ + + + + + + Xiaomi Smart Speaker + + + + + + +
+

Home Assistant instances found:

+
+
+ + + + \ No newline at end of file diff --git a/api/utils.py b/api/utils.py new file mode 100644 index 0000000..0a3118c --- /dev/null +++ b/api/utils.py @@ -0,0 +1,11 @@ +import socket +import fcntl +import struct + +def get_ip_address(ifname): + s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + return socket.inet_ntoa(fcntl.ioctl( + s.fileno(), + 0x8915, # SIOCGIFADDR + struct.pack('256s', ifname[:15].encode('utf-8')) + )[20:24])