#!/usr/bin/env python3 """ Termux SMS API Server Bridges SMS Campaign Manager (Ubuntu) with Termux API (Android) This server runs on the Android device in Termux and provides REST API endpoints for the main SMS Campaign Manager to send messages using native Android SMS capabilities instead of ADB automation. """ from flask import Flask, request, jsonify import subprocess import json import time import logging import hmac import hashlib import os from datetime import datetime from typing import Dict, List, Optional, Any # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('/data/data/com.termux/files/home/logs/sms-api.log'), logging.StreamHandler() ] ) logger = logging.getLogger(__name__) app = Flask(__name__) # Configuration CONFIG = { 'SECRET_KEY': os.environ.get('SMS_API_SECRET', 'termux-sms-campaign-2025'), 'MAX_MESSAGE_LENGTH': 160, 'RATE_LIMIT_DELAY': 2.0, # Seconds between messages 'ALLOWED_COMMANDS': [ 'termux-sms-send', 'termux-sms-list', 'termux-battery-status', 'termux-location', 'termux-notification' ] } class SMSApiServer: """Main SMS API server class""" def __init__(self): self.last_send_time = 0 self.message_count = 0 self.start_time = time.time() def authenticate_request(self, request_data: str, signature: str) -> bool: """Verify HMAC signature for request authentication""" try: expected_signature = hmac.new( CONFIG['SECRET_KEY'].encode(), request_data.encode(), hashlib.sha256 ).hexdigest() return hmac.compare_digest(signature, expected_signature) except Exception as e: logger.error(f"Authentication error: {e}") return False def execute_termux_command(self, command: List[str]) -> Dict[str, Any]: """Execute Termux API command with error handling""" if not command or command[0] not in CONFIG['ALLOWED_COMMANDS']: return {'success': False, 'error': 'Command not allowed'} try: logger.info(f"Executing: {' '.join(command)}") result = subprocess.run( command, capture_output=True, text=True, timeout=30 ) return { 'success': result.returncode == 0, 'stdout': result.stdout.strip(), 'stderr': result.stderr.strip(), 'return_code': result.returncode } except subprocess.TimeoutExpired: return {'success': False, 'error': 'Command timeout'} except Exception as e: return {'success': False, 'error': str(e)} def rate_limit_check(self) -> bool: """Check if enough time has passed since last message""" current_time = time.time() if current_time - self.last_send_time < CONFIG['RATE_LIMIT_DELAY']: return False self.last_send_time = current_time return True def send_sms(self, phone: str, message: str) -> Dict[str, Any]: """Send SMS using Termux API""" # Input validation if not phone or not message: return {'success': False, 'error': 'Phone and message required'} if len(message) > CONFIG['MAX_MESSAGE_LENGTH']: return {'success': False, 'error': f'Message too long (max {CONFIG["MAX_MESSAGE_LENGTH"]} chars)'} # Rate limiting if not self.rate_limit_check(): return {'success': False, 'error': 'Rate limit exceeded, please wait'} # Execute SMS send command command = ['termux-sms-send', '-n', phone, message] result = self.execute_termux_command(command) if result['success']: self.message_count += 1 logger.info(f"SMS sent to {phone}: {message[:50]}...") # Send confirmation notification self.execute_termux_command([ 'termux-notification', '--title', 'SMS Sent', '--content', f'Message sent to {phone}' ]) return { 'success': result['success'], 'error': result.get('error') or result.get('stderr'), 'timestamp': datetime.now().isoformat(), 'phone': phone, 'message_length': len(message), 'total_sent': self.message_count } # Global server instance sms_server = SMSApiServer() # API Endpoints # Web interface route @app.route("/") def index(): """Web interface for SMS API Server""" from flask import render_template_string return render_template_string(""" SMS API Server - Android

๐Ÿ“ฑ SMS API Server

Android Termux Interface

๐Ÿ  Homelab๐Ÿ“Š Monitor

๐Ÿš€ Running on Android (Termux)

โœ… Server Status: Operational

Device IP: {{ device_ip }}

Port: 5001

Environment: Termux on Android

๐Ÿ”— API Endpoints

๐Ÿ“Š Health Check

GET /health

Returns server status, uptime, and message statistics

๐Ÿ“ฑ Send SMS

POST /api/sms/send

Send SMS messages with name substitution support

๐Ÿ”‹ Battery Status

GET /api/device/battery

Get real-time Android device battery information

๐Ÿ“ Location

GET /api/device/location

Get GPS coordinates (with permissions)

โ„น๏ธ Device Info

GET /api/device/info

System information and device details

๐Ÿงช Quick Tests

๐Ÿ“Š Health Check ๐Ÿ”‹ Battery โ„น๏ธ Device Info

๐Ÿ  Part of SMS Campaign Manager System

๐Ÿ–ฅ๏ธ Main Interface: http://{{ homelab_ip }}:5000

""", device_ip="10.0.0.193", homelab_ip="10.0.0.190") @app.route('/health', methods=['GET']) def health_check(): """Health check endpoint""" uptime = time.time() - sms_server.start_time return jsonify({ 'status': 'healthy', 'timestamp': datetime.now().isoformat(), 'uptime_seconds': int(uptime), 'messages_sent': sms_server.message_count, 'version': '1.0.0' }) @app.route('/api/sms/send', methods=['POST']) def send_sms(): """Send SMS message via Termux API""" try: data = request.get_json() if not data: return jsonify({'success': False, 'error': 'JSON data required'}), 400 # Extract parameters phone = data.get('phone', '').strip() message = data.get('message', '').strip() name = data.get('name', '') # Message template substitution (like existing ui.sh) if name and '{name}' in message: message = message.replace('{name}', name) # Send SMS result = sms_server.send_sms(phone, message) status_code = 200 if result['success'] else 400 return jsonify(result), status_code except Exception as e: logger.error(f"SMS send error: {e}") return jsonify({'success': False, 'error': str(e)}), 500 @app.route('/api/sms/list', methods=['GET']) def list_sms(): """List SMS messages""" try: limit = request.args.get('limit', '10') offset = request.args.get('offset', '0') command = ['termux-sms-list', '-l', limit, '-o', offset] result = sms_server.execute_termux_command(command) if result['success']: try: sms_data = json.loads(result['stdout']) if result['stdout'] else [] return jsonify({ 'success': True, 'messages': sms_data, 'count': len(sms_data) }) except json.JSONDecodeError: return jsonify({ 'success': True, 'messages': result['stdout'], 'raw_output': True }) else: return jsonify({'success': False, 'error': result['error']}), 400 except Exception as e: logger.error(f"SMS list error: {e}") return jsonify({'success': False, 'error': str(e)}), 500 @app.route('/api/device/battery', methods=['GET']) def get_battery_status(): """Get device battery status""" try: result = sms_server.execute_termux_command(['termux-battery-status']) if result['success']: battery_data = json.loads(result['stdout']) return jsonify({ 'success': True, 'battery': battery_data, 'timestamp': datetime.now().isoformat() }) else: return jsonify({'success': False, 'error': result['error']}), 400 except Exception as e: logger.error(f"Battery status error: {e}") return jsonify({'success': False, 'error': str(e)}), 500 @app.route('/api/device/location', methods=['GET']) def get_location(): """Get device GPS location""" try: result = sms_server.execute_termux_command(['termux-location']) if result['success'] and result['stdout']: try: location_data = json.loads(result['stdout']) return jsonify({ 'success': True, 'location': location_data, 'timestamp': datetime.now().isoformat() }) except json.JSONDecodeError: return jsonify({ 'success': True, 'location': result['stdout'], 'raw_output': True }) else: return jsonify({'success': False, 'error': 'Location unavailable'}), 400 except Exception as e: logger.error(f"Location error: {e}") return jsonify({'success': False, 'error': str(e)}), 500 @app.route('/api/device/info', methods=['GET']) def get_device_info(): """Get general device information""" try: # Get multiple device stats battery_result = sms_server.execute_termux_command(['termux-battery-status']) info = { 'timestamp': datetime.now().isoformat(), 'uptime': time.time() - sms_server.start_time, 'messages_sent': sms_server.message_count, 'api_version': '1.0.0', 'termux_api_available': True } if battery_result['success']: try: battery_data = json.loads(battery_result['stdout']) info['battery'] = battery_data except json.JSONDecodeError: pass return jsonify({'success': True, 'device_info': info}) except Exception as e: logger.error(f"Device info error: {e}") return jsonify({'success': False, 'error': str(e)}), 500 @app.route('/api/campaign/notify', methods=['POST']) def campaign_notification(): """Send notification about campaign status""" try: data = request.get_json() title = data.get('title', 'SMS Campaign') message = data.get('message', 'Campaign update') result = sms_server.execute_termux_command([ 'termux-notification', '--title', title, '--content', message ]) return jsonify({ 'success': result['success'], 'error': result.get('error') }) except Exception as e: logger.error(f"Notification error: {e}") return jsonify({'success': False, 'error': str(e)}), 500 if __name__ == '__main__': # Create logs directory os.makedirs('/data/data/com.termux/files/home/logs', exist_ok=True) logger.info("Starting Termux SMS API Server...") logger.info(f"Available commands: {CONFIG['ALLOWED_COMMANDS']}") # Get local IP for display try: ip_result = subprocess.run([ 'ifconfig', '2>/dev/null', '|', 'grep', '-A1', 'wlan0', '|', 'grep', 'inet', '|', 'awk', '{print $2}', '|', 'cut', '-d:', '-f2' ], shell=True, capture_output=True, text=True) local_ip = ip_result.stdout.strip() or '10.0.0.193' except: local_ip = '10.0.0.193' print(f""" ๐Ÿš€ Termux SMS API Server Starting ๐Ÿ“ฑ Device IP: {local_ip} ๐ŸŒ API Base URL: http://{local_ip}:5001 ๐Ÿ”— Health Check: http://{local_ip}:5001/health ๐Ÿ“ž SMS Endpoint: http://{local_ip}:5001/api/sms/send ๐Ÿ”‹ Battery API: http://{local_ip}:5001/api/device/battery Access from Ubuntu homelab: curl http://{local_ip}:5001/health """) app.run(host='0.0.0.0', port=5001, debug=False)