"""One-number SessionTalk SMS bridge. See README.md before exposing this service."""
import base64
from contextlib import contextmanager
import hashlib
import hmac
import json
import os
import re
import sqlite3
import sys
import time
import uuid
from datetime import datetime, timezone
from urllib.parse import urlparse

import requests
from cryptography.fernet import Fernet
from flask import Flask, abort, jsonify, request
from werkzeug.exceptions import HTTPException
from nacl.signing import VerifyKey
from twilio.request_validator import RequestValidator


def create_app(settings=None):
    cfg = dict(os.environ if settings is None else settings)
    for key in ('MAILBOX_ID', 'MAILBOX_TOKEN', 'SMS_NUMBER', 'DATABASE_KEY', 'PUBLIC_URL', 'PROVIDER'):
        if not cfg.get(key):
            raise ValueError('Missing setting: ' + key)
    uuid.UUID(cfg['MAILBOX_ID'])
    if not re.fullmatch(r'[A-Za-z0-9_-]{43}', cfg['MAILBOX_TOKEN']):
        raise ValueError('MAILBOX_TOKEN must be 32 random bytes, base64url without padding')
    if not re.fullmatch(r'\+[1-9][0-9]{6,14}', cfg['SMS_NUMBER']):
        raise ValueError('SMS_NUMBER must be international E.164')
    base = urlparse(cfg['PUBLIC_URL'])
    if base.scheme != 'https' or not base.hostname or base.username or base.query or base.fragment or base.path not in ('', '/'):
        raise ValueError('PUBLIC_URL must be an HTTPS origin')
    provider = cfg['PROVIDER']
    required = {'aa': ('AA_PASSWORD', 'WEBHOOK_SECRET'),
                'twilio': ('TWILIO_ACCOUNT_SID', 'TWILIO_AUTH_TOKEN'),
                'telnyx': ('TELNYX_API_KEY', 'TELNYX_PUBLIC_KEY'),
                'custom': ('CUSTOM_SEND_URL', 'CUSTOM_API_KEY', 'WEBHOOK_SECRET')}
    if provider not in required:
        raise ValueError('Unknown provider')
    for key in required[provider]:
        if not cfg.get(key):
            raise ValueError('Missing setting: ' + key)
    if provider in ('aa', 'custom') and not re.fullmatch(r'[A-Za-z0-9_-]{43}', cfg['WEBHOOK_SECRET']):
        raise ValueError('WEBHOOK_SECRET must be a separate random 32-byte token')
    cipher = Fernet(cfg['DATABASE_KEY'].encode())
    app = Flask(__name__)
    # Framework error logs can include the secret A&A request path.
    app.logger.disabled = True
    app.json.ensure_ascii = False

    @app.errorhandler(Exception)
    def safe_error(error):
        status = error.code if isinstance(error, HTTPException) else 400 if isinstance(error, (KeyError, TypeError, ValueError)) else 503
        return jsonify(error='Request rejected' if status < 500 else 'Service unavailable'), status

    app.config['MAX_CONTENT_LENGTH'] = 96 * 1024
    app.config['BRIDGE'] = cfg
    database = cfg.get('DATABASE', '/data/sms.sqlite3')
    allowed_hosts = set(cfg.get('PUSH_HOSTS', 'eupush.sessiontalk.co.uk,uspush.sessiontalk.co.uk,asiapush.sessiontalk.co.uk,pushtest.sessiontalk.co.uk').split(','))

    @contextmanager
    def connect():
        db = sqlite3.connect(database, timeout=10)
        db.row_factory = sqlite3.Row
        db.execute('PRAGMA journal_mode=WAL')
        db.execute('PRAGMA synchronous=FULL')
        try:
            with db:
                yield db
        finally:
            db.close()

    with connect() as db:
        db.executescript('''
          CREATE TABLE IF NOT EXISTS messages(seq INTEGER PRIMARY KEY AUTOINCREMENT, id TEXT UNIQUE NOT NULL, payload BLOB NOT NULL);
          CREATE TABLE IF NOT EXISTS devices(id TEXT PRIMARY KEY, route BLOB NOT NULL, expires INTEGER NOT NULL);
          CREATE TABLE IF NOT EXISTS jobs(device TEXT, message TEXT, due INTEGER, attempts INTEGER DEFAULT 0,
            done INTEGER DEFAULT 0, PRIMARY KEY(device,message));
        ''')

    def pack(data):
        return cipher.encrypt(json.dumps(data, ensure_ascii=False).encode())

    def unpack(data):
        return json.loads(cipher.decrypt(data))

    def authenticated():
        if not hmac.compare_digest(request.headers.get('Authorization', ''), 'Bearer ' + cfg['MAILBOX_TOKEN']):
            abort(401)

    def store(event_id, sender, destination, text):
        if destination != cfg['SMS_NUMBER']:
            abort(403)
        if not isinstance(sender, str) or not re.fullmatch(r'(?:\+[1-9][0-9]{6,14}|[A-Za-z0-9][A-Za-z0-9 ._-]{0,19})', sender):
            abort(400)
        if not isinstance(text, str) or not text or len(text.encode()) > 65536 or not event_id or len(event_id) > 512:
            abort(400)
        mid = str(uuid.uuid5(uuid.UUID(cfg['MAILBOX_ID']), provider + ':' + event_id))
        message = {'id': mid, 'peer': sender, 'body': text, 'direction': 'inbound',
                   'state': 'received', 'created_at': datetime.now(timezone.utc).isoformat()}
        with connect() as db:
            inserted = db.execute('INSERT OR IGNORE INTO messages(id,payload) VALUES(?,?)', (mid, pack(message))).rowcount
            if inserted:
                db.execute('INSERT OR IGNORE INTO jobs(device,message,due) SELECT id,?,? FROM devices WHERE expires>?', (mid, int(time.time()), int(time.time())))
        # Commit the message and jobs before acknowledging the provider.
        return mid

    @app.get('/inbox')
    def inbox():
        authenticated()
        after = request.args.get('after', '0')
        if not re.fullmatch(r'[0-9]{1,18}', after):
            abort(400)
        with connect() as db:
            rows = db.execute('SELECT * FROM messages WHERE seq>? ORDER BY seq LIMIT 51', (int(after),)).fetchall()
        page = rows[:50]
        return jsonify(messages=[unpack(r['payload']) for r in page], next=page[-1]['seq'] if page else int(after), more=len(rows) > 50)

    @app.get('/inbox/<mid>')
    def message(mid):
        authenticated()
        with connect() as db:
            row = db.execute('SELECT payload FROM messages WHERE id=?', (mid,)).fetchone()
        if row is None:
            abort(404)
        return jsonify(message=unpack(row['payload']))

    @app.put('/devices')
    def report():
        authenticated()
        data = request.get_json()
        if not isinstance(data, dict):
            abort(400)
        route = urlparse(str(data.get('pushUrl', '')))
        if route.scheme != 'https' or route.hostname not in allowed_hosts or route.netloc != route.hostname or route.path != '/byo/webservice/push' or route.query or route.fragment:
            abort(400)
        if data.get('mailboxId') != cfg['MAILBOX_ID'] or not isinstance(data.get('profileId'), int) or data['profileId'] <= 0:
            abort(400)
        if not isinstance(data.get('deviceId'), str) or not 1 <= len(data['deviceId']) <= 200 or not isinstance(data.get('pushToken'), str) or not 1 <= len(data['pushToken']) <= 2048:
            abort(400)
        if not isinstance(data.get('expiresAt'), int) or not time.time() < data['expiresAt'] <= time.time() + 604800:
            abort(400)
        key = hashlib.sha256(json.dumps([data['deviceId'], data['profileId']]).encode()).hexdigest()
        with connect() as db:
            db.execute('INSERT INTO devices(id,route,expires) VALUES(?,?,?) ON CONFLICT(id) DO UPDATE SET route=excluded.route,expires=excluded.expires', (key, pack(data), data['expiresAt']))
        return jsonify(ok=True)

    @app.post('/webhooks/aa/<secret>')
    def aa(secret):
        if provider != 'aa' or not hmac.compare_digest(secret, cfg.get('WEBHOOK_SECRET', '')):
            abort(403)
        form = request.form
        stamp = form.get('scts', '')
        if not stamp:
            abort(400)
        # A&A does not document a signed callback or a unique delivery ID.
        # A stable digest makes its identical callback retries idempotent.
        event = hashlib.sha256(json.dumps([form.get(k, '') for k in ('oa', 'da', 'scts', 'ud', 'udh')]).encode()).hexdigest()
        store(event, form.get('oa', ''), form.get('da', ''), form.get('ud', ''))
        return '', 204

    @app.post('/webhooks/twilio')
    def twilio():
        if provider != 'twilio' or request.query_string:
            abort(403)
        if not RequestValidator(cfg['TWILIO_AUTH_TOKEN']).validate(cfg['PUBLIC_URL'].rstrip('/') + '/webhooks/twilio', request.form, request.headers.get('X-Twilio-Signature', '')):
            abort(403)
        if request.form.get('AccountSid') != cfg['TWILIO_ACCOUNT_SID']:
            abort(403)
        store(request.form.get('MessageSid', ''), request.form.get('From', ''), request.form.get('To', ''), request.form.get('Body', ''))
        return '<Response/>', 200, {'Content-Type': 'application/xml'}

    @app.post('/webhooks/telnyx')
    def telnyx():
        if provider != 'telnyx':
            abort(403)
        stamp = request.headers.get('telnyx-timestamp', '')
        raw = request.get_data()
        try:
            if abs(time.time() - int(stamp)) > 300:
                raise ValueError()
            VerifyKey(base64.b64decode(cfg['TELNYX_PUBLIC_KEY'], validate=True)).verify(stamp.encode() + b'|' + raw, base64.b64decode(request.headers.get('telnyx-signature-ed25519', ''), validate=True))
        except Exception:
            abort(403)
        data = request.get_json()['data']
        if data.get('event_type') != 'message.received':
            return '', 204
        payload = data['payload']
        targets = payload.get('to', [])
        if not any(t.get('phone_number') == cfg['SMS_NUMBER'] for t in targets):
            abort(403)
        store(payload['id'], payload['from']['phone_number'], cfg['SMS_NUMBER'], payload.get('text', ''))
        return '', 204

    @app.post('/webhooks/custom')
    def custom():
        if provider != 'custom':
            abort(403)
        stamp = request.headers.get('X-SMS-Timestamp', '')
        raw = request.get_data()
        signature = hmac.new(cfg['WEBHOOK_SECRET'].encode(), stamp.encode() + b'.' + raw, hashlib.sha256).hexdigest()
        try:
            valid = abs(time.time() - int(stamp)) <= 300
        except ValueError:
            valid = False
        if not valid or not hmac.compare_digest(signature, request.headers.get('X-SMS-Signature', '')):
            abort(403)
        data = request.get_json()
        store(data.get('id', ''), data.get('from', ''), data.get('to', ''), data.get('text', ''))
        return '', 204

    @app.post('/send')
    def send():
        # Matches the app's Custom four-field form; never use SIP credentials.
        if not hmac.compare_digest(request.form.get('username', ''), cfg['MAILBOX_ID']) or not hmac.compare_digest(request.form.get('password', ''), cfg['MAILBOX_TOKEN']):
            abort(401)
        to, text = request.form.get('da', ''), request.form.get('ud', '')
        if not re.fullmatch(r'\+[1-9][0-9]{6,14}', to) or not 1 <= len(text) <= 1600:
            abort(400)
        try:
            if provider == 'aa':
                url, kwargs = 'https://sms.aa.net.uk/sms.cgi', {'data': {'username': cfg['SMS_NUMBER'], 'password': cfg['AA_PASSWORD'], 'da': to, 'ud': text}}
            elif provider == 'twilio':
                url = 'https://api.twilio.com/2010-04-01/Accounts/' + cfg['TWILIO_ACCOUNT_SID'] + '/Messages.json'
                kwargs = {'auth': (cfg['TWILIO_ACCOUNT_SID'], cfg['TWILIO_AUTH_TOKEN']), 'data': {'From': cfg['SMS_NUMBER'], 'To': to, 'Body': text}}
            else:
                url = 'https://api.telnyx.com/v2/messages' if provider == 'telnyx' else cfg['CUSTOM_SEND_URL']
                kwargs = {'headers': {'Authorization': 'Bearer ' + cfg['TELNYX_API_KEY' if provider == 'telnyx' else 'CUSTOM_API_KEY']}, 'json': {'from': cfg['SMS_NUMBER'], 'to': to, 'text': text}}
            if urlparse(url).scheme != 'https' or urlparse(url).username:
                abort(500)
            with requests.post(url, timeout=15, allow_redirects=False, stream=True, **kwargs) as response:
                if not 200 <= response.status_code < 300:
                    return 'ERR: Provider rejected request', 502
                if provider == 'aa' and not response.raw.read(3).startswith(b'OK:'):
                    return 'ERR: Provider rejected request', 502
        except requests.RequestException:
            return 'ERR: Delivery unconfirmed; check provider before retrying', 502
        return 'OK: Accepted', 200

    def work_once():
        now = int(time.time())
        with connect() as db:
            rows = db.execute('SELECT j.*, d.route, m.payload FROM jobs j JOIN devices d ON d.id=j.device JOIN messages m ON m.id=j.message WHERE j.done=0 AND j.due<=? AND d.expires>? ORDER BY j.due LIMIT 50', (now, now)).fetchall()
        for row in rows:
            route, msg = unpack(row['route']), unpack(row['payload'])
            done = False
            try:
                response = requests.post(route['pushUrl'], headers={'Authorization': 'Bearer ' + route['pushToken']}, json={'messageId': msg['id'], 'remoteNumber': msg['peer'], 'text': msg['body'][:80]}, timeout=15, allow_redirects=False)
                done = response.status_code == 200 and response.json().get('ok') is True
                response.close()
            except (requests.RequestException, ValueError):
                pass  # No credential-bearing URL, payload or exception logging.
            with connect() as db:
                db.execute('UPDATE jobs SET done=?,attempts=attempts+1,due=? WHERE device=? AND message=?', (int(done), now + min(3600, 5 * 2**min(row['attempts'], 10)), row['device'], row['message']))
        return len(rows)

    app.work_once = work_once
    return app


if __name__ == '__main__':
    if sys.argv[1:] != ['worker']:
        raise SystemExit('Run HTTP with gunicorn "bridge:create_app()"; run the queue with python bridge.py worker')
    application = create_app()
    while True:
        application.work_once()
        time.sleep(2)