admin/campaign_connector
1
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

updates

Browse Source
This commit is contained in:
bunker-admin 2026-02-27 15:01:44 -07:00
parent 81fa6c983d
commit 2457662e12
1 changed files with 145 additions and 217 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

362
android/termux-sms-api-server.py
View File

@ -6,6 +6,9 @@ Bridges SMS Campaign Manager (Ubuntu) with Termux API (Android)
This server runs on the Android device in Termux and provides REST API This server runs on the Android device in Termux and provides REST API
endpoints for the main SMS Campaign Manager to send messages using endpoints for the main SMS Campaign Manager to send messages using
native Android SMS capabilities instead of ADB automation. native Android SMS capabilities instead of ADB automation.
All endpoints require API key authentication via X-API-Key header.
Localhost requests (127.0.0.1, ::1) are exempt for watchdog health checks.
""" """
from flask import Flask, request, jsonify from flask import Flask, request, jsonify
@ -52,12 +55,12 @@ CONFIG = {
class SMSApiServer: class SMSApiServer:
"""Main SMS API server class""" """Main SMS API server class"""
def __init__(self): def __init__(self):
self.last_send_time = 0 self.last_send_time = 0
self.message_count = 0 self.message_count = 0
self.start_time = time.time() self.start_time = time.time()
def authenticate_request(self, request_data: str, signature: str) -> bool: def authenticate_request(self, request_data: str, signature: str) -> bool:
"""Verify HMAC signature for request authentication""" """Verify HMAC signature for request authentication"""
try: try:
@ -70,21 +73,21 @@ class SMSApiServer:
except Exception as e: except Exception as e:
logger.error(f"Authentication error: {e}") logger.error(f"Authentication error: {e}")
return False return False
def execute_termux_command(self, command: List[str]) -> Dict[str, Any]: def execute_termux_command(self, command: List[str]) -> Dict[str, Any]:
"""Execute Termux API command with error handling""" """Execute Termux API command with error handling"""
if not command or command[0] not in CONFIG['ALLOWED_COMMANDS']: if not command or command[0] not in CONFIG['ALLOWED_COMMANDS']:
return {'success': False, 'error': 'Command not allowed'} return {'success': False, 'error': 'Command not allowed'}
try: try:
logger.info(f"Executing: {' '.join(command)}") logger.info(f"Executing: {' '.join(command)}")
result = subprocess.run( result = subprocess.run(
command, command,
capture_output=True, capture_output=True,
text=True, text=True,
timeout=30 timeout=30
) )
return { return {
'success': result.returncode == 0, 'success': result.returncode == 0,
'stdout': result.stdout.strip(), 'stdout': result.stdout.strip(),
@ -95,27 +98,27 @@ class SMSApiServer:
return {'success': False, 'error': 'Command timeout'} return {'success': False, 'error': 'Command timeout'}
except Exception as e: except Exception as e:
return {'success': False, 'error': str(e)} return {'success': False, 'error': str(e)}
def get_sms_history(self, phone: Optional[str] = None, limit: int = 100) -> Dict[str, Any]: 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""" """Get SMS history for a specific phone number or all messages"""
try: try:
command = ['termux-sms-list'] command = ['termux-sms-list']
if limit: if limit:
command.extend(['-l', str(limit)]) command.extend(['-l', str(limit)])
result = self.execute_termux_command(command) result = self.execute_termux_command(command)
if result['success'] and result['stdout']: if result['success'] and result['stdout']:
try: try:
messages = json.loads(result['stdout']) messages = json.loads(result['stdout'])
# Filter by phone number if specified # Filter by phone number if specified
if phone: if phone:
# Clean phone number for comparison # Clean phone number for comparison
clean_phone = phone.replace('+', '').replace('-', '').replace(' ', '').replace('(', '').replace(')', '') clean_phone = phone.replace('+', '').replace('-', '').replace(' ', '').replace('(', '').replace(')', '')
messages = [msg for msg in messages messages = [msg for msg in messages
if msg.get('number', '').replace('+', '').replace('-', '').replace(' ', '').replace('(', '').replace(')', '') == clean_phone] if msg.get('number', '').replace('+', '').replace('-', '').replace(' ', '').replace('(', '').replace(')', '') == clean_phone]
return { return {
'success': True, 'success': True,
'messages': messages, 'messages': messages,
@ -123,23 +126,23 @@ class SMSApiServer:
} }
except json.JSONDecodeError: except json.JSONDecodeError:
return {'success': False, 'error': 'Failed to parse SMS data'} return {'success': False, 'error': 'Failed to parse SMS data'}
return {'success': False, 'error': 'Failed to retrieve SMS history'} return {'success': False, 'error': 'Failed to retrieve SMS history'}
except Exception as e: except Exception as e:
logger.error(f"Error getting SMS history: {e}") logger.error(f"Error getting SMS history: {e}")
return {'success': False, 'error': str(e)} return {'success': False, 'error': str(e)}
def get_contact_name(self, phone: str) -> Optional[str]: def get_contact_name(self, phone: str) -> Optional[str]:
"""Get contact name from phone's contact list""" """Get contact name from phone's contact list"""
try: try:
# Use termux-contact-list command if available # Use termux-contact-list command if available
result = self.execute_termux_command(['termux-contact-list']) result = self.execute_termux_command(['termux-contact-list'])
if result['success'] and result['stdout']: if result['success'] and result['stdout']:
try: try:
contacts = json.loads(result['stdout']) contacts = json.loads(result['stdout'])
clean_phone = phone.replace('+', '').replace('-', '').replace(' ', '').replace('(', '').replace(')', '') clean_phone = phone.replace('+', '').replace('-', '').replace(' ', '').replace('(', '').replace(')', '')
for contact in contacts: for contact in contacts:
if 'phoneNumbers' in contact: if 'phoneNumbers' in contact:
for phone_entry in contact['phoneNumbers']: for phone_entry in contact['phoneNumbers']:
@ -148,12 +151,12 @@ class SMSApiServer:
return contact.get('name') return contact.get('name')
except json.JSONDecodeError: except json.JSONDecodeError:
pass pass
return None return None
except Exception as e: except Exception as e:
logger.error(f"Error getting contact name for {phone}: {e}") logger.error(f"Error getting contact name for {phone}: {e}")
return None return None
def rate_limit_check(self) -> bool: def rate_limit_check(self) -> bool:
"""Check if enough time has passed since last message""" """Check if enough time has passed since last message"""
current_time = time.time() current_time = time.time()
@ -161,7 +164,7 @@ class SMSApiServer:
return False return False
self.last_send_time = current_time self.last_send_time = current_time
return True return True
def send_sms(self, phone: str, message: str) -> Dict[str, Any]: def send_sms(self, phone: str, message: str) -> Dict[str, Any]:
"""Send SMS using Termux API""" """Send SMS using Termux API"""
# Input validation # Input validation
@ -169,32 +172,32 @@ class SMSApiServer:
error_msg = 'Phone and message required' error_msg = 'Phone and message required'
logger.error(f"SMS validation failed: {error_msg}") logger.error(f"SMS validation failed: {error_msg}")
return {'success': False, 'error': error_msg} return {'success': False, 'error': error_msg}
# Clean phone number # Clean phone number
clean_phone = phone.replace('+', '').replace('-', '').replace(' ', '').replace('(', '').replace(')', '') clean_phone = phone.replace('+', '').replace('-', '').replace(' ', '').replace('(', '').replace(')', '')
if len(message) > CONFIG['MAX_MESSAGE_LENGTH']: if len(message) > CONFIG['MAX_MESSAGE_LENGTH']:
error_msg = f'Message too long ({len(message)} chars, max {CONFIG["MAX_MESSAGE_LENGTH"]})' error_msg = f'Message too long ({len(message)} chars, max {CONFIG["MAX_MESSAGE_LENGTH"]})'
logger.error(f"SMS validation failed: {error_msg}") logger.error(f"SMS validation failed: {error_msg}")
return {'success': False, 'error': error_msg} return {'success': False, 'error': error_msg}
# Rate limiting # Rate limiting
if not self.rate_limit_check(): if not self.rate_limit_check():
error_msg = 'Rate limit exceeded, please wait' error_msg = 'Rate limit exceeded, please wait'
logger.warning(f"SMS rate limited for {phone}") logger.warning(f"SMS rate limited for {phone}")
return {'success': False, 'error': error_msg} return {'success': False, 'error': error_msg}
# Log the SMS attempt # Log the SMS attempt
logger.info(f"Attempting to send SMS to {phone} (length: {len(message)} chars)") logger.info(f"Attempting to send SMS to {phone} (length: {len(message)} chars)")
# Execute SMS send command # Execute SMS send command
command = ['termux-sms-send', '-n', clean_phone, message] command = ['termux-sms-send', '-n', clean_phone, message]
result = self.execute_termux_command(command) result = self.execute_termux_command(command)
if result['success']: if result['success']:
self.message_count += 1 self.message_count += 1
logger.info(f"✅ SMS sent successfully to {phone}: {message[:50]}...") logger.info(f"SMS sent successfully to {phone}: {message[:50]}...")
# Send confirmation notification # Send confirmation notification
self.execute_termux_command([ self.execute_termux_command([
'termux-notification', 'termux-notification',
@ -202,8 +205,8 @@ class SMSApiServer:
'--content', f'Message sent to {phone}' '--content', f'Message sent to {phone}'
]) ])
else: else:
logger.error(f"❌ SMS send failed to {phone}: {result.get('error', 'Unknown error')}") logger.error(f"SMS send failed to {phone}: {result.get('error', 'Unknown error')}")
return { return {
'success': result['success'], 'success': result['success'],
'error': result.get('error') or result.get('stderr'), 'error': result.get('error') or result.get('stderr'),
@ -222,136 +225,40 @@ def verify_api_key():
expected_key = CONFIG['SECRET_KEY'] expected_key = CONFIG['SECRET_KEY']
if not api_key: if not api_key:
logger.warning(f"⚠️ No API key provided for {request.path} from {request.remote_addr}")
return False return False
if not hmac.compare_digest(api_key, expected_key): if not hmac.compare_digest(api_key, expected_key):
logger.warning(f"⚠️ Invalid API key for {request.path} from {request.remote_addr}")
return False return False
logger.info(f"✅ Valid API key for {request.path}")
return True return True
# ---------------------------------------------------------------------------
# Global auth: every request from a remote client must provide a valid API key.
# Localhost (127.0.0.1 / ::1) is exempt so the watchdog health check works.
# ---------------------------------------------------------------------------
@app.before_request
def require_api_key():
"""Enforce API key authentication on all endpoints for remote clients."""
if request.remote_addr in ('127.0.0.1', '::1'):
return None # allow localhost (watchdog health checks)
if not verify_api_key():
logger.warning(f"Auth failed: {request.method} {request.path} from {request.remote_addr}")
return jsonify({
'success': False,
'error': 'Authentication required',
'message': 'Please provide valid API key via X-API-Key header'
}), 401
return None
# API Endpoints # 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="endpoint">
<h3>📇 Contact List</h3>
<p><code>GET /api/contacts/list</code></p>
<p>Fetch all contacts from phone (with optional search)</p>
</div>
<div class="endpoint">
<h3>🧪 Test Contact Structure</h3>
<p><code>GET /api/contacts/test</code></p>
<p>Analyze contact list JSON structure and fields</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>
<a href="/api/contacts/test">📇 Test Contacts</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']) @app.route('/health', methods=['GET'])
def health_check(): def health_check():
"""Health check endpoint""" """Health check endpoint"""
uptime = time.time() - sms_server.start_time uptime = time.time() - sms_server.start_time
return jsonify({ return jsonify({
'status': 'healthy', 'status': 'healthy',
'timestamp': datetime.now().isoformat(), 'timestamp': datetime.now().isoformat(),
@ -363,47 +270,39 @@ def health_check():
@app.route('/api/sms/send', methods=['POST']) @app.route('/api/sms/send', methods=['POST'])
def send_sms(): def send_sms():
"""Send SMS message via Termux API""" """Send SMS message via Termux API"""
# Verify API key
if not verify_api_key():
return jsonify({
'success': False,
'error': 'Authentication required',
'message': 'Please provide valid API key via X-API-Key header'
}), 401
try: try:
data = request.get_json() data = request.get_json()
if not data: if not data:
logger.error("SMS send endpoint: No JSON data provided") logger.error("SMS send endpoint: No JSON data provided")
return jsonify({'success': False, 'error': 'JSON data required'}), 400 return jsonify({'success': False, 'error': 'JSON data required'}), 400
# Extract parameters # Extract parameters
phone = data.get('phone', '').strip() phone = data.get('phone', '').strip()
message = data.get('message', '').strip() message = data.get('message', '').strip()
name = data.get('name', '') name = data.get('name', '')
logger.info(f"SMS send request: phone={phone}, name={name}, message_length={len(message) if message else 0}") logger.info(f"SMS send request: phone={phone}, name={name}, message_length={len(message) if message else 0}")
# Validate required fields # Validate required fields
if not phone: if not phone:
logger.error("SMS send validation: Missing phone number") logger.error("SMS send validation: Missing phone number")
return jsonify({'success': False, 'error': 'Phone number required'}), 400 return jsonify({'success': False, 'error': 'Phone number required'}), 400
if not message: if not message:
logger.error("SMS send validation: Missing message") logger.error("SMS send validation: Missing message")
return jsonify({'success': False, 'error': 'Message required'}), 400 return jsonify({'success': False, 'error': 'Message required'}), 400
# Message template substitution (like existing ui.sh) # Message template substitution (like existing ui.sh)
if name and '{name}' in message: if name and '{name}' in message:
message = message.replace('{name}', name) message = message.replace('{name}', name)
# Send SMS # Send SMS
result = sms_server.send_sms(phone, message) result = sms_server.send_sms(phone, message)
status_code = 200 if result['success'] else 400 status_code = 200 if result['success'] else 400
logger.info(f"SMS send result: success={result['success']}, error={result.get('error', 'None')}") logger.info(f"SMS send result: success={result['success']}, error={result.get('error', 'None')}")
return jsonify(result), status_code return jsonify(result), status_code
except Exception as e: except Exception as e:
logger.error(f"SMS send endpoint error: {e}") logger.error(f"SMS send endpoint error: {e}")
return jsonify({'success': False, 'error': str(e)}), 500 return jsonify({'success': False, 'error': str(e)}), 500
@ -414,10 +313,10 @@ def list_sms():
try: try:
limit = request.args.get('limit', '10') limit = request.args.get('limit', '10')
offset = request.args.get('offset', '0') offset = request.args.get('offset', '0')
command = ['termux-sms-list', '-l', limit, '-o', offset] command = ['termux-sms-list', '-l', limit, '-o', offset]
result = sms_server.execute_termux_command(command) result = sms_server.execute_termux_command(command)
if result['success']: if result['success']:
try: try:
sms_data = json.loads(result['stdout']) if result['stdout'] else [] sms_data = json.loads(result['stdout']) if result['stdout'] else []
@ -434,7 +333,7 @@ def list_sms():
}) })
else: else:
return jsonify({'success': False, 'error': result['error']}), 400 return jsonify({'success': False, 'error': result['error']}), 400
except Exception as e: except Exception as e:
logger.error(f"SMS list error: {e}") logger.error(f"SMS list error: {e}")
return jsonify({'success': False, 'error': str(e)}), 500 return jsonify({'success': False, 'error': str(e)}), 500
@ -444,7 +343,7 @@ def get_battery_status():
"""Get device battery status""" """Get device battery status"""
try: try:
result = sms_server.execute_termux_command(['termux-battery-status']) result = sms_server.execute_termux_command(['termux-battery-status'])
if result['success']: if result['success']:
battery_data = json.loads(result['stdout']) battery_data = json.loads(result['stdout'])
return jsonify({ return jsonify({
@ -454,7 +353,7 @@ def get_battery_status():
}) })
else: else:
return jsonify({'success': False, 'error': result['error']}), 400 return jsonify({'success': False, 'error': result['error']}), 400
except Exception as e: except Exception as e:
logger.error(f"Battery status error: {e}") logger.error(f"Battery status error: {e}")
return jsonify({'success': False, 'error': str(e)}), 500 return jsonify({'success': False, 'error': str(e)}), 500
@ -464,7 +363,7 @@ def get_location():
"""Get device GPS location""" """Get device GPS location"""
try: try:
result = sms_server.execute_termux_command(['termux-location']) result = sms_server.execute_termux_command(['termux-location'])
if result['success'] and result['stdout']: if result['success'] and result['stdout']:
try: try:
location_data = json.loads(result['stdout']) location_data = json.loads(result['stdout'])
@ -481,7 +380,7 @@ def get_location():
}) })
else: else:
return jsonify({'success': False, 'error': 'Location unavailable'}), 400 return jsonify({'success': False, 'error': 'Location unavailable'}), 400
except Exception as e: except Exception as e:
logger.error(f"Location error: {e}") logger.error(f"Location error: {e}")
return jsonify({'success': False, 'error': str(e)}), 500 return jsonify({'success': False, 'error': str(e)}), 500
@ -492,7 +391,7 @@ def get_device_info():
try: try:
# Get multiple device stats # Get multiple device stats
battery_result = sms_server.execute_termux_command(['termux-battery-status']) battery_result = sms_server.execute_termux_command(['termux-battery-status'])
info = { info = {
'timestamp': datetime.now().isoformat(), 'timestamp': datetime.now().isoformat(),
'uptime': time.time() - sms_server.start_time, 'uptime': time.time() - sms_server.start_time,
@ -500,16 +399,16 @@ def get_device_info():
'api_version': '1.0.0', 'api_version': '1.0.0',
'termux_api_available': True 'termux_api_available': True
} }
if battery_result['success']: if battery_result['success']:
try: try:
battery_data = json.loads(battery_result['stdout']) battery_data = json.loads(battery_result['stdout'])
info['battery'] = battery_data info['battery'] = battery_data
except json.JSONDecodeError: except json.JSONDecodeError:
pass pass
return jsonify({'success': True, 'device_info': info}) return jsonify({'success': True, 'device_info': info})
except Exception as e: except Exception as e:
logger.error(f"Device info error: {e}") logger.error(f"Device info error: {e}")
return jsonify({'success': False, 'error': str(e)}), 500 return jsonify({'success': False, 'error': str(e)}), 500
@ -520,14 +419,14 @@ def get_sms_history():
try: try:
phone = request.args.get('phone') phone = request.args.get('phone')
limit = request.args.get('limit', 100, type=int) limit = request.args.get('limit', 100, type=int)
result = sms_server.get_sms_history(phone, limit) result = sms_server.get_sms_history(phone, limit)
if result['success']: if result['success']:
return jsonify(result) return jsonify(result)
else: else:
return jsonify(result), 400 return jsonify(result), 400
except Exception as e: except Exception as e:
logger.error(f"SMS history error: {e}") logger.error(f"SMS history error: {e}")
return jsonify({'success': False, 'error': str(e)}), 500 return jsonify({'success': False, 'error': str(e)}), 500
@ -538,12 +437,12 @@ def get_sms_inbox():
try: try:
since = request.args.get('since', type=int) since = request.args.get('since', type=int)
limit = request.args.get('limit', 100, type=int) limit = request.args.get('limit', 100, type=int)
result = sms_server.get_sms_history(None, limit) result = sms_server.get_sms_history(None, limit)
if result['success']: if result['success']:
messages = result['messages'] messages = result['messages']
# Filter by timestamp if 'since' parameter provided # Filter by timestamp if 'since' parameter provided
if since: if since:
filtered_messages = [] filtered_messages = []
@ -557,7 +456,7 @@ def get_sms_inbox():
# If timestamp parsing fails, include the message # If timestamp parsing fails, include the message
filtered_messages.append(msg) filtered_messages.append(msg)
messages = filtered_messages messages = filtered_messages
return jsonify({ return jsonify({
'success': True, 'success': True,
'messages': messages, 'messages': messages,
@ -565,7 +464,7 @@ def get_sms_inbox():
}) })
else: else:
return jsonify(result), 400 return jsonify(result), 400
except Exception as e: except Exception as e:
logger.error(f"SMS inbox error: {e}") logger.error(f"SMS inbox error: {e}")
return jsonify({'success': False, 'error': str(e)}), 500 return jsonify({'success': False, 'error': str(e)}), 500
@ -575,14 +474,14 @@ def get_contact_info(phone):
"""Get contact information for a phone number""" """Get contact information for a phone number"""
try: try:
contact_name = sms_server.get_contact_name(phone) contact_name = sms_server.get_contact_name(phone)
return jsonify({ return jsonify({
'success': True, 'success': True,
'phone': phone, 'phone': phone,
'name': contact_name, 'name': contact_name,
'has_name': contact_name is not None 'has_name': contact_name is not None
}) })
except Exception as e: except Exception as e:
logger.error(f"Contact info error: {e}") logger.error(f"Contact info error: {e}")
return jsonify({'success': False, 'error': str(e)}), 500 return jsonify({'success': False, 'error': str(e)}), 500
@ -590,41 +489,33 @@ def get_contact_info(phone):
@app.route('/api/sms/send-reply', methods=['POST']) @app.route('/api/sms/send-reply', methods=['POST'])
def send_reply(): def send_reply():
"""Send a reply message with enhanced tracking""" """Send a reply message with enhanced tracking"""
# Verify API key
if not verify_api_key():
return jsonify({
'success': False,
'error': 'Authentication required',
'message': 'Please provide valid API key via X-API-Key header'
}), 401
try: try:
data = request.get_json() data = request.get_json()
if not data: if not data:
return jsonify({'success': False, 'error': 'No JSON data provided'}), 400 return jsonify({'success': False, 'error': 'No JSON data provided'}), 400
phone = data.get('phone') phone = data.get('phone')
message = data.get('message') message = data.get('message')
conversation_id = data.get('conversation_id') conversation_id = data.get('conversation_id')
if not phone or not message: if not phone or not message:
return jsonify({'success': False, 'error': 'Phone and message required'}), 400 return jsonify({'success': False, 'error': 'Phone and message required'}), 400
# Rate limiting check # Rate limiting check
if not sms_server.rate_limit_check(): if not sms_server.rate_limit_check():
return jsonify({ return jsonify({
'success': False, 'success': False,
'error': f'Rate limited. Wait {CONFIG["RATE_LIMIT_DELAY"]} seconds between messages' 'error': f'Rate limited. Wait {CONFIG["RATE_LIMIT_DELAY"]} seconds between messages'
}), 429 }), 429
# Send via termux-sms-send # Send via termux-sms-send
command = ['termux-sms-send', '-n', phone, message] command = ['termux-sms-send', '-n', phone, message]
result = sms_server.execute_termux_command(command) result = sms_server.execute_termux_command(command)
if result['success']: if result['success']:
sms_server.message_count += 1 sms_server.message_count += 1
return jsonify({ return jsonify({
'success': True, 'success': True,
'message': 'Reply sent successfully', 'message': 'Reply sent successfully',
@ -638,7 +529,7 @@ def send_reply():
'success': False, 'success': False,
'error': f'Failed to send SMS: {result.get("stderr", "Unknown error")}' 'error': f'Failed to send SMS: {result.get("stderr", "Unknown error")}'
}), 500 }), 500
except Exception as e: except Exception as e:
logger.error(f"Send reply error: {e}") logger.error(f"Send reply error: {e}")
return jsonify({'success': False, 'error': str(e)}), 500 return jsonify({'success': False, 'error': str(e)}), 500
@ -666,6 +557,45 @@ def campaign_notification():
logger.error(f"Notification error: {e}") logger.error(f"Notification error: {e}")
return jsonify({'success': False, 'error': str(e)}), 500 return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/logs/tail', methods=['GET'])
def tail_logs():
"""Get last N lines from the SMS API log file"""
try:
lines_param = request.args.get('lines', '100', type=str)
try:
num_lines = min(500, max(1, int(lines_param)))
except (ValueError, TypeError):
num_lines = 100
log_path = os.path.expanduser('~/logs/sms-api.log')
if not os.path.isfile(log_path):
return jsonify({
'success': True,
'lines': [],
'total_lines': 0,
'file_size': 0
})
file_size = os.path.getsize(log_path)
with open(log_path, 'r', encoding='utf-8', errors='replace') as f:
all_lines = f.readlines()
total_lines = len(all_lines)
tail_lines = [line.rstrip('\n') for line in all_lines[-num_lines:]]
return jsonify({
'success': True,
'lines': tail_lines,
'total_lines': total_lines,
'file_size': file_size
})
except Exception as e:
logger.error(f"Log tail error: {e}")
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/contacts/test', methods=['GET']) @app.route('/api/contacts/test', methods=['GET'])
def test_contacts(): def test_contacts():
"""Test endpoint to fetch and examine raw contact list structure""" """Test endpoint to fetch and examine raw contact list structure"""
@ -772,7 +702,7 @@ if __name__ == '__main__':
# Validate required configuration # Validate required configuration
if not CONFIG['SECRET_KEY']: if not CONFIG['SECRET_KEY']:
logger.critical("❌ SECURITY ERROR: SMS_API_SECRET or TERMUX_API_KEY environment variable is required!") logger.critical("SECURITY ERROR: SMS_API_SECRET or TERMUX_API_KEY environment variable is required!")
logger.critical("Set SMS_API_SECRET environment variable before starting the server") logger.critical("Set SMS_API_SECRET environment variable before starting the server")
logger.critical("Generate a secure key with: python -c \"import secrets; print(secrets.token_hex(32))\"") logger.critical("Generate a secure key with: python -c \"import secrets; print(secrets.token_hex(32))\"")
print("\n" + "="*80) print("\n" + "="*80)
@ -788,9 +718,9 @@ if __name__ == '__main__':
exit(1) exit(1)
logger.info("Starting Termux SMS API Server...") logger.info("Starting Termux SMS API Server...")
logger.info(f"✅ API authentication configured") logger.info("API authentication configured (all remote endpoints protected)")
logger.info(f"Available commands: {CONFIG['ALLOWED_COMMANDS']}") logger.info(f"Available commands: {CONFIG['ALLOWED_COMMANDS']}")
# Get local IP for display (secure method without shell=True) # Get local IP for display (secure method without shell=True)
try: try:
import socket import socket
@ -801,18 +731,16 @@ if __name__ == '__main__':
s.close() s.close()
except Exception as e: except Exception as e:
logger.warning(f"Could not determine IP address: {e}") logger.warning(f"Could not determine IP address: {e}")
local_ip = '10.0.0.193' local_ip = '0.0.0.0'
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: print(f"""
curl http://{local_ip}:5001/health Termux SMS API Server Starting
Device IP: {local_ip}
API Base URL: http://{local_ip}:5001
All endpoints require X-API-Key header (except localhost)
Access from server:
curl -H "X-API-Key: $SMS_API_SECRET" http://{local_ip}:5001/health
""") """)
app.run(host='0.0.0.0', port=5001, debug=False) app.run(host='0.0.0.0', port=5001, debug=False)