campaign_connector/android/termux-sms-api-server.py
2025-08-25 09:41:16 -06:00

623 lines
24 KiB
Python
Raw Blame History

This file contains invisible Unicode characters

This file contains invisible Unicode characters that are indistinguishable to humans but may be processed differently by a computer. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

#!/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 get_sms_history(self, phone: Optional[str] = None, limit: int = 100) -> Dict[str, Any]:
"""Get SMS history for a specific phone number or all messages"""
try:
command = ['termux-sms-list']
if limit:
command.extend(['-l', str(limit)])
result = self.execute_termux_command(command)
if result['success'] and result['stdout']:
try:
messages = json.loads(result['stdout'])
# Filter by phone number if specified
if phone:
# Clean phone number for comparison
clean_phone = phone.replace('+', '').replace('-', '').replace(' ', '').replace('(', '').replace(')', '')
messages = [msg for msg in messages
if msg.get('number', '').replace('+', '').replace('-', '').replace(' ', '').replace('(', '').replace(')', '') == clean_phone]
return {
'success': True,
'messages': messages,
'count': len(messages)
}
except json.JSONDecodeError:
return {'success': False, 'error': 'Failed to parse SMS data'}
return {'success': False, 'error': 'Failed to retrieve SMS history'}
except Exception as e:
logger.error(f"Error getting SMS history: {e}")
return {'success': False, 'error': str(e)}
def get_contact_name(self, phone: str) -> Optional[str]:
"""Get contact name from phone's contact list"""
try:
# Use termux-contact-list command if available
result = self.execute_termux_command(['termux-contact-list'])
if result['success'] and result['stdout']:
try:
contacts = json.loads(result['stdout'])
clean_phone = phone.replace('+', '').replace('-', '').replace(' ', '').replace('(', '').replace(')', '')
for contact in contacts:
if 'phoneNumbers' in contact:
for phone_entry in contact['phoneNumbers']:
contact_phone = phone_entry.get('number', '').replace('+', '').replace('-', '').replace(' ', '').replace('(', '').replace(')', '')
if contact_phone == clean_phone:
return contact.get('name')
except json.JSONDecodeError:
pass
return None
except Exception as e:
logger.error(f"Error getting contact name for {phone}: {e}")
return None
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("""
<html>
<head>
<title>SMS API Server - Android</title>
<meta name="viewport" content="width=device-width, initial-scale=1">
<style>
body { font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, sans-serif;
margin: 0; padding: 20px; background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
min-height: 100vh; color: white; }
.container { max-width: 800px; margin: 0 auto; background: rgba(255,255,255,0.1);
padding: 30px; border-radius: 15px; backdrop-filter: blur(10px); box-shadow: 0 8px 32px rgba(0,0,0,0.1); }
h1 { text-align: center; margin-bottom: 30px; font-size: 2.5em; text-shadow: 2px 2px 4px rgba(0,0,0,0.3); }
.status { background: rgba(0,255,0,0.2); padding: 20px; border-radius: 10px; margin: 20px 0;
border-left: 5px solid #00ff00; }
.endpoint { background: rgba(255,255,255,0.1); padding: 15px; margin: 10px 0; border-radius: 8px;
border-left: 3px solid #fff; }
.endpoint h3 { margin: 0 0 10px 0; color: #fff; }
.endpoint code { background: rgba(0,0,0,0.3); padding: 5px 10px; border-radius: 5px;
font-family: "Courier New", monospace; }
.endpoint p { margin: 5px 0; opacity: 0.9; }
.test-links { text-align: center; margin: 20px 0; }
.test-links a { display: inline-block; margin: 5px 10px; padding: 10px 20px;
background: rgba(255,255,255,0.2); color: white; text-decoration: none;
border-radius: 25px; transition: all 0.3s ease; }
.test-links a:hover { background: rgba(255,255,255,0.3); transform: translateY(-2px); }
.nav-links a:hover { background: rgba(255,255,255,0.3) !important; transform: translateY(-1px); }
.footer a:hover { opacity: 0.8; }
.footer { text-align: center; margin-top: 40px; opacity: 0.7; font-size: 0.9em; }
</style>
</head>
<body>
<div class="container">
<div class="header-nav" style="display: flex; justify-content: space-between; align-items: center; margin-bottom: 30px; padding-bottom: 20px; border-bottom: 1px solid rgba(255,255,255,0.2);"><div><h1 style="margin: 0; font-size: 2.5em;">📱 SMS API Server</h1><p style="margin: 0; opacity: 0.8; font-size: 1.1em;">Android Termux Interface</p></div><div class="nav-links" style="display: flex; gap: 15px;"><a href="http://localhost:5000/" style="background: rgba(255,255,255,0.2); padding: 8px 16px; border-radius: 20px; color: white; text-decoration: none; font-size: 0.9em; transition: all 0.3s ease;">🏠 Homelab</a><a href="http://10.0.0.193:5000" style="background: rgba(255,255,255,0.2); padding: 8px 16px; border-radius: 20px; color: white; text-decoration: none; font-size: 0.9em; transition: all 0.3s ease;">📊 Monitor</a></div></div>
<h2>🚀 Running on Android (Termux)</h2>
<div class="status">
<h3>✅ Server Status: Operational</h3>
<p><strong>Device IP:</strong> {{ device_ip }}</p>
<p><strong>Port:</strong> 5001</p>
<p><strong>Environment:</strong> Termux on Android</p>
</div>
<h3>🔗 API Endpoints</h3>
<div class="endpoint">
<h3>📊 Health Check</h3>
<p><code>GET /health</code></p>
<p>Returns server status, uptime, and message statistics</p>
</div>
<div class="endpoint">
<h3>📱 Send SMS</h3>
<p><code>POST /api/sms/send</code></p>
<p>Send SMS messages with name substitution support</p>
</div>
<div class="endpoint">
<h3>🔋 Battery Status</h3>
<p><code>GET /api/device/battery</code></p>
<p>Get real-time Android device battery information</p>
</div>
<div class="endpoint">
<h3>📍 Location</h3>
<p><code>GET /api/device/location</code></p>
<p>Get GPS coordinates (with permissions)</p>
</div>
<div class="endpoint">
<h3> Device Info</h3>
<p><code>GET /api/device/info</code></p>
<p>System information and device details</p>
</div>
<div class="test-links">
<h3>🧪 Quick Tests</h3>
<a href="/health">📊 Health Check</a>
<a href="/api/device/battery">🔋 Battery</a>
<a href="/api/device/info"> Device Info</a>
</div>
<div class="footer" style="text-align: center; margin-top: 40px; padding-top: 20px; border-top: 1px solid rgba(255,255,255,0.2); opacity: 0.7; font-size: 0.9em;">
<p>📱 SMS Campaign Manager • Android SMS API Service</p>
<div style="margin-top: 10px;">
<a href="http://localhost:5000/" style="color: #fff; margin: 0 10px; text-decoration: none;">🏠 Homelab Dashboard</a> |
<a href="http://10.0.0.193:5000" style="color: #fff; margin: 0 10px; text-decoration: none;">📊 Android Monitor</a> |
<span style="font-size: 0.8em;">© 2025 Campaign System</span>
</div>
</div>
</div>
</body>
</html>
""", 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/sms/history', methods=['GET'])
def get_sms_history():
"""Get SMS history for conversation sync"""
try:
phone = request.args.get('phone')
limit = request.args.get('limit', 100, type=int)
result = sms_server.get_sms_history(phone, limit)
if result['success']:
return jsonify(result)
else:
return jsonify(result), 400
except Exception as e:
logger.error(f"SMS history error: {e}")
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/sms/inbox', methods=['GET'])
def get_sms_inbox():
"""Get SMS inbox messages (compatibility endpoint)"""
try:
since = request.args.get('since', type=int)
limit = request.args.get('limit', 100, type=int)
result = sms_server.get_sms_history(None, limit)
if result['success']:
messages = result['messages']
# Filter by timestamp if 'since' parameter provided
if since:
filtered_messages = []
for msg in messages:
# Convert received timestamp to compare with since
try:
msg_time = datetime.strptime(msg.get('received', ''), '%Y-%m-%d %H:%M:%S').timestamp()
if msg_time > since:
filtered_messages.append(msg)
except:
# If timestamp parsing fails, include the message
filtered_messages.append(msg)
messages = filtered_messages
return jsonify({
'success': True,
'messages': messages,
'count': len(messages)
})
else:
return jsonify(result), 400
except Exception as e:
logger.error(f"SMS inbox error: {e}")
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/contact/<phone>', methods=['GET'])
def get_contact_info(phone):
"""Get contact information for a phone number"""
try:
contact_name = sms_server.get_contact_name(phone)
return jsonify({
'success': True,
'phone': phone,
'name': contact_name,
'has_name': contact_name is not None
})
except Exception as e:
logger.error(f"Contact info error: {e}")
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/sms/send-reply', methods=['POST'])
def send_reply():
"""Send a reply message with enhanced tracking"""
try:
data = request.get_json()
if not data:
return jsonify({'success': False, 'error': 'No JSON data provided'}), 400
phone = data.get('phone')
message = data.get('message')
conversation_id = data.get('conversation_id')
if not phone or not message:
return jsonify({'success': False, 'error': 'Phone and message required'}), 400
# Rate limiting check
if not sms_server.rate_limit_check():
return jsonify({
'success': False,
'error': f'Rate limited. Wait {CONFIG["RATE_LIMIT_DELAY"]} seconds between messages'
}), 429
# Send via termux-sms-send
command = ['termux-sms-send', '-n', phone, message]
result = sms_server.execute_termux_command(command)
if result['success']:
sms_server.message_count += 1
return jsonify({
'success': True,
'message': 'Reply sent successfully',
'conversation_id': conversation_id,
'timestamp': datetime.now().isoformat(),
'phone': phone,
'message_sent': message
})
else:
return jsonify({
'success': False,
'error': f'Failed to send SMS: {result.get("stderr", "Unknown error")}'
}), 500
except Exception as e:
logger.error(f"Send reply 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)