"""GazdaTárs local production server with versioned SQLite backups.""" from __future__ import annotations import hashlib import json import os import sqlite3 import base64 import re import smtplib from datetime import datetime, timedelta, timezone from email.message import EmailMessage from http import HTTPStatus from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from urllib.parse import parse_qs, urlparse ROOT = Path(__file__).resolve().parent DB_PATH = Path(os.environ.get("GAZDATARS_DB", ROOT / "data" / "gazdatars.db")) # A legfeljebb 4 MiB-os XLSX base64 kódolva nagyobb JSON-kérést eredményez. MAX_BODY = 6 * 1024 * 1024 REGISTRATION_SENDER = os.environ.get("REGISTRATION_FROM", os.environ.get("SMTP_FROM", "gazdatars@gazdatars.com")).strip() REGISTRATION_RECIPIENT = os.environ.get("REGISTRATION_RECIPIENT", "gazdatars@gazdatars.com").strip() ALLOWED_ORIGIN = os.environ.get("CORS_ALLOWED_ORIGIN", "https://gazdatars.com").strip() SENSITIVE_BACKUP_KEYS = frozenset(("gazdatars_proba_felhasznalo", "gazdatars_bejelentkezve", "gazdatars_device_id", "gazdatars_anon_id")) _BACKUPS_SCRUBBED = False def now_iso() -> str: return datetime.now(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z") def sanitize_backup_payload(payload): if not isinstance(payload, dict): return payload return {key: value for key, value in payload.items() if key not in SENSITIVE_BACKUP_KEYS} def connect() -> sqlite3.Connection: global _BACKUPS_SCRUBBED DB_PATH.parent.mkdir(parents=True, exist_ok=True) db = sqlite3.connect(DB_PATH) db.row_factory = sqlite3.Row db.execute("PRAGMA journal_mode=WAL") db.execute("PRAGMA foreign_keys=ON") db.executescript(""" CREATE TABLE IF NOT EXISTS users ( user_id TEXT PRIMARY KEY, created_at TEXT NOT NULL, updated_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS backups ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id TEXT NOT NULL REFERENCES users(user_id) ON DELETE CASCADE, created_at TEXT NOT NULL, reason TEXT NOT NULL, checksum TEXT NOT NULL, payload TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS backups_user_created ON backups(user_id, id DESC); CREATE TABLE IF NOT EXISTS email_deliveries ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id TEXT NOT NULL REFERENCES users(user_id) ON DELETE CASCADE, created_at TEXT NOT NULL, recipient_name TEXT, recipient_email TEXT NOT NULL, filename TEXT NOT NULL, backup_id INTEGER REFERENCES backups(id), status TEXT NOT NULL, error_message TEXT ); CREATE INDEX IF NOT EXISTS deliveries_user_created ON email_deliveries(user_id, id DESC); CREATE TABLE IF NOT EXISTS subscriptions ( user_id TEXT PRIMARY KEY REFERENCES users(user_id) ON DELETE CASCADE, updated_at TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'incomplete', plan_code TEXT NOT NULL DEFAULT 'monthly_900_huf_gross', gross_amount INTEGER NOT NULL DEFAULT 900, billing_name TEXT NOT NULL, billing_country TEXT NOT NULL DEFAULT 'Magyarország', billing_postcode TEXT NOT NULL, billing_city TEXT NOT NULL, billing_address TEXT NOT NULL, tax_number TEXT, billing_email TEXT NOT NULL, provider_customer_id TEXT, provider_subscription_id TEXT ); """) subscription_columns = {row[1] for row in db.execute("PRAGMA table_info(subscriptions)")} if "cancelled_at" not in subscription_columns: db.execute("ALTER TABLE subscriptions ADD COLUMN cancelled_at TEXT") user_columns = {row[1] for row in db.execute("PRAGMA table_info(users)")} if "trial_started_at" not in user_columns: db.execute("ALTER TABLE users ADD COLUMN trial_started_at TEXT") if "trial_ends_at" not in user_columns: db.execute("ALTER TABLE users ADD COLUMN trial_ends_at TEXT") if "registration_email_sent_at" not in user_columns: db.execute("ALTER TABLE users ADD COLUMN registration_email_sent_at TEXT") if "registration_email_error" not in user_columns: db.execute("ALTER TABLE users ADD COLUMN registration_email_error TEXT") if not _BACKUPS_SCRUBBED: for row in db.execute("SELECT id, payload FROM backups").fetchall(): try: original = json.loads(row["payload"]) except (TypeError, json.JSONDecodeError): continue cleaned = sanitize_backup_payload(original) if cleaned != original: encoded = json.dumps(cleaned, ensure_ascii=False, sort_keys=True, separators=(",", ":")) checksum = hashlib.sha256(encoded.encode("utf-8")).hexdigest() db.execute("UPDATE backups SET payload=?, checksum=? WHERE id=?", (encoded, checksum, row["id"])) db.commit() _BACKUPS_SCRUBBED = True return db def subscription_access(db: sqlite3.Connection, user_id: str) -> dict: user = db.execute("SELECT trial_started_at, trial_ends_at FROM users WHERE user_id=?", (user_id,)).fetchone() subscription = db.execute("SELECT status FROM subscriptions WHERE user_id=?", (user_id,)).fetchone() active = bool(subscription and subscription["status"] == "active") trial_end = user["trial_ends_at"] if user else None trial_active = bool(trial_end and trial_end > now_iso()) return {"allowed": active or trial_active, "trialActive": trial_active, "trialEndsAt": trial_end, "subscriptionActive": active} class Handler(SimpleHTTPRequestHandler): def __init__(self, *args, **kwargs): super().__init__(*args, directory=str(ROOT), **kwargs) def send_json(self, value, status=HTTPStatus.OK): body = json.dumps(value, ensure_ascii=False).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-store") self.send_cors_headers() self.end_headers() self.wfile.write(body) def send_cors_headers(self): origin = self.headers.get("Origin", "") if origin and origin == ALLOWED_ORIGIN: self.send_header("Access-Control-Allow-Origin", origin) self.send_header("Vary", "Origin") self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS") self.send_header("Access-Control-Allow-Headers", "Content-Type") def do_OPTIONS(self): if not urlparse(self.path).path.startswith("/api/"): return self.send_error(HTTPStatus.NOT_FOUND) if self.headers.get("Origin", "") != ALLOWED_ORIGIN: return self.send_json({"error": "Nem engedélyezett eredet."}, HTTPStatus.FORBIDDEN) self.send_response(HTTPStatus.NO_CONTENT) self.send_cors_headers() self.send_header("Content-Length", "0") self.end_headers() def read_json(self): length = int(self.headers.get("Content-Length", "0")) if length <= 0 or length > MAX_BODY: raise ValueError("Érvénytelen kérésméret.") return json.loads(self.rfile.read(length)) @staticmethod def valid_user(value): return isinstance(value, str) and 1 <= len(value) <= 200 def do_GET(self): parsed = urlparse(self.path) if parsed.path == "/api/health": return self.send_json({"ok": True, "service": "gazdatars-api"}) if parsed.path == "/api/backups": user_id = parse_qs(parsed.query).get("user", [""])[0] if not self.valid_user(user_id): return self.send_json({"error": "Hiányzó felhasználó."}, HTTPStatus.BAD_REQUEST) with connect() as db: rows = db.execute( "SELECT id, created_at, reason FROM backups WHERE user_id=? ORDER BY id DESC LIMIT 50", (user_id,), ).fetchall() latest = db.execute( "SELECT payload FROM backups WHERE user_id=? ORDER BY id DESC LIMIT 1", (user_id,) ).fetchone() return self.send_json({"backups": [dict(r) for r in rows], "current": sanitize_backup_payload(json.loads(latest[0])) if latest else None}) if parsed.path == "/api/history": user_id = parse_qs(parsed.query).get("user", [""])[0] if not self.valid_user(user_id): return self.send_json({"error": "Hiányzó felhasználó."}, HTTPStatus.BAD_REQUEST) with connect() as db: backups = db.execute( "SELECT id, created_at, reason FROM backups WHERE user_id=? ORDER BY id DESC LIMIT 100", (user_id,) ).fetchall() deliveries = db.execute( "SELECT id, created_at, recipient_name, recipient_email, filename, backup_id, status, error_message FROM email_deliveries WHERE user_id=? ORDER BY id DESC LIMIT 100", (user_id,), ).fetchall() return self.send_json({"backups": [dict(r) for r in backups], "deliveries": [dict(r) for r in deliveries]}) if parsed.path == "/api/subscription": user_id = parse_qs(parsed.query).get("user", [""])[0] if not self.valid_user(user_id): return self.send_json({"error": "Hiányzó felhasználó."}, HTTPStatus.BAD_REQUEST) with connect() as db: row = db.execute("SELECT * FROM subscriptions WHERE user_id=?", (user_id,)).fetchone() access = subscription_access(db, user_id) return self.send_json({"subscription": dict(row) if row else None, "access": access, "price": {"amount": 900, "currency": "HUF", "gross": True, "interval": "month"}}) return super().do_GET() def do_POST(self): parsed = urlparse(self.path) try: data = self.read_json() user_id = data.get("userId", "") if not self.valid_user(user_id): raise ValueError("Hiányzó felhasználó.") if parsed.path == "/api/register": return self.register_trial(user_id) if parsed.path == "/api/backups": return self.create_backup(user_id, data) if parsed.path == "/api/backups/restore": return self.restore_backup(user_id, data) if parsed.path == "/api/email/falugazdasz": return self.send_to_adviser(user_id, data) if parsed.path == "/api/subscription/billing": return self.save_billing(user_id, data) if parsed.path == "/api/subscription/checkout": return self.start_checkout(user_id) if parsed.path == "/api/subscription/cancel": return self.cancel_subscription(user_id, data) self.send_json({"error": "Ismeretlen végpont."}, HTTPStatus.NOT_FOUND) except (ValueError, TypeError, json.JSONDecodeError) as exc: self.send_json({"error": str(exc)}, HTTPStatus.BAD_REQUEST) def create_backup(self, user_id, data): payload = sanitize_backup_payload(data.get("payload")) if not isinstance(payload, dict): raise ValueError("A mentendő adat hibás.") encoded = json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")) checksum = hashlib.sha256(encoded.encode("utf-8")).hexdigest() timestamp = now_iso() with connect() as db: db.execute("INSERT OR IGNORE INTO users (user_id,created_at,updated_at) VALUES (?, ?, ?)", (user_id, timestamp, timestamp)) access = subscription_access(db, user_id) if not access["allowed"]: return self.send_json({"error": "A 30 napos próbaidőszak lejárt. A központi mentéshez előfizetés szükséges.", "trialExpired": True}, HTTPStatus.PAYMENT_REQUIRED) latest = db.execute( "SELECT id, created_at, checksum, payload FROM backups WHERE user_id=? ORDER BY id DESC LIMIT 1", (user_id,) ).fetchone() base_id = data.get("baseBackupId") if latest and str(base_id or "") != str(latest["id"]): return self.send_json({ "error": "Az adatokat egy másik eszköz is módosította.", "conflict": True, "latestId": latest["id"], "current": json.loads(latest["payload"]), }, HTTPStatus.CONFLICT) if latest and latest["checksum"] == checksum: return self.send_json({"ok": True, "id": latest["id"], "savedAt": latest["created_at"], "unchanged": True}) cur = db.execute( "INSERT INTO backups(user_id, created_at, reason, checksum, payload) VALUES (?, ?, ?, ?, ?)", (user_id, timestamp, str(data.get("reason", "automatikus mentés"))[:80], checksum, encoded), ) db.execute("UPDATE users SET updated_at=? WHERE user_id=?", (timestamp, user_id)) self.send_json({"ok": True, "id": cur.lastrowid, "savedAt": timestamp}, HTTPStatus.CREATED) def register_trial(self, user_id): if not re.fullmatch(r"[^@\s]+@[^@\s]+\.[^@\s]+", user_id): raise ValueError("A regisztrációs e-mail-cím érvénytelen.") timestamp = now_iso() trial_end = (datetime.now(timezone.utc) + timedelta(days=30)).isoformat(timespec="seconds").replace("+00:00", "Z") with connect() as db: row = db.execute("SELECT trial_started_at, trial_ends_at, registration_email_sent_at FROM users WHERE user_id=?", (user_id,)).fetchone() existing = bool(row and row["trial_started_at"]) if existing: timestamp = row["trial_started_at"] trial_end = row["trial_ends_at"] if row: if not existing: db.execute("UPDATE users SET updated_at=?, trial_started_at=?, trial_ends_at=? WHERE user_id=?", (timestamp, timestamp, trial_end, user_id)) else: db.execute("INSERT INTO users (user_id,created_at,updated_at,trial_started_at,trial_ends_at) VALUES (?,?,?,?,?)", (user_id,timestamp,timestamp,timestamp,trial_end)) email_sent = bool(row and row["registration_email_sent_at"]) if not email_sent: email_sent = self.send_registration_emails(user_id, trial_end) sent_at = now_iso() if email_sent else None error = None if email_sent else "A regisztrációs e-mail küldése sikertelen vagy nincs beállítva." with connect() as db: db.execute( "UPDATE users SET registration_email_sent_at=?, registration_email_error=? WHERE user_id=?", (sent_at, error, user_id), ) self.send_json({ "ok": True, "trialStartedAt": timestamp, "trialEndsAt": trial_end, "existing": existing, "registrationEmailSent": email_sent, }, HTTPStatus.OK if existing else HTTPStatus.CREATED) @staticmethod def send_registration_emails(user_id, trial_end): host = os.environ.get("SMTP_HOST") if not host or not REGISTRATION_SENDER or not REGISTRATION_RECIPIENT: print("Registration e-mail skipped: SMTP is not configured.") return False confirmation = EmailMessage() confirmation["From"] = REGISTRATION_SENDER confirmation["To"] = user_id confirmation["Reply-To"] = REGISTRATION_RECIPIENT confirmation["Subject"] = "GazdaTárs – sikeres regisztráció" confirmation.set_content( "Kedves Gazdálkodó!\n\n" "A GazdaTárs-regisztrációd sikeresen elkészült. " f"A 30 napos próbaidőszak vége: {trial_end}.\n\n" "Ha kérdésed van, írj a gazdatars@gazdatars.com címre.\n\n" "Üdvözlettel:\nGazdaTárs" ) notification = EmailMessage() notification["From"] = REGISTRATION_SENDER notification["To"] = REGISTRATION_RECIPIENT notification["Reply-To"] = user_id notification["Subject"] = "GazdaTárs – új regisztráció" notification.set_content( "Új GazdaTárs-regisztráció érkezett.\n\n" f"Felhasználó e-mail-címe: {user_id}\n" f"Regisztráció időpontja: {now_iso()}\n" f"Próbaidőszak vége: {trial_end}" ) port = int(os.environ.get("SMTP_PORT", "465" if os.environ.get("SMTP_SSL") == "1" else "587")) smtp_class = smtplib.SMTP_SSL if os.environ.get("SMTP_SSL") == "1" else smtplib.SMTP try: with smtp_class(host, port, timeout=20) as smtp: if smtp_class is smtplib.SMTP and os.environ.get("SMTP_STARTTLS", "1") == "1": smtp.starttls() if os.environ.get("SMTP_USER"): smtp.login(os.environ["SMTP_USER"], os.environ.get("SMTP_PASSWORD", "")) smtp.send_message(confirmation) smtp.send_message(notification) except (OSError, smtplib.SMTPException) as exc: print(f"Registration e-mail sending failed: {exc}") return False return True def save_billing(self, user_id, data): required = ("billingName", "billingPostcode", "billingCity", "billingAddress", "billingEmail") values = {key: str(data.get(key, "")).strip() for key in required} if any(not values[key] for key in required): raise ValueError("Minden kötelező számlázási adatot tölts ki.") if not re.fullmatch(r"[^@\s]+@[^@\s]+\.[^@\s]+", values["billingEmail"]): raise ValueError("A számlázási e-mail-cím érvénytelen.") timestamp = now_iso() with connect() as db: db.execute("INSERT OR IGNORE INTO users (user_id,created_at,updated_at) VALUES (?, ?, ?)", (user_id, timestamp, timestamp)) db.execute(""" INSERT INTO subscriptions (user_id,updated_at,status,billing_name,billing_country,billing_postcode,billing_city,billing_address,tax_number,billing_email) VALUES (?,?,?,?,?,?,?,?,?,?) ON CONFLICT(user_id) DO UPDATE SET updated_at=excluded.updated_at,billing_name=excluded.billing_name, billing_country=excluded.billing_country,billing_postcode=excluded.billing_postcode,billing_city=excluded.billing_city, billing_address=excluded.billing_address,tax_number=excluded.tax_number,billing_email=excluded.billing_email """, (user_id,timestamp,"incomplete",values["billingName"],str(data.get("billingCountry","Magyarország")).strip() or "Magyarország", values["billingPostcode"],values["billingCity"],values["billingAddress"],str(data.get("taxNumber","")).strip(),values["billingEmail"])) self.send_json({"ok": True, "savedAt": timestamp}) def start_checkout(self, user_id): checkout_url = os.environ.get("PAYMENT_CHECKOUT_URL", "").strip() if not checkout_url: return self.send_json({"error": "Az online fizetési szolgáltató még nincs beállítva."}, HTTPStatus.SERVICE_UNAVAILABLE) with connect() as db: if not db.execute("SELECT 1 FROM subscriptions WHERE user_id=?", (user_id,)).fetchone(): raise ValueError("Előbb add meg és mentsd el a számlázási adatokat.") separator = "&" if "?" in checkout_url else "?" self.send_json({"ok": True, "checkoutUrl": checkout_url + separator + "user=" + user_id}) def cancel_subscription(self, user_id, data): if data.get("confirmed") is not True: raise ValueError("A lemondáshoz külön megerősítés szükséges.") timestamp = now_iso() with connect() as db: row = db.execute("SELECT provider_subscription_id FROM subscriptions WHERE user_id=?", (user_id,)).fetchone() if not row: return self.send_json({"error": "Nem található lemondható előfizetés."}, HTTPStatus.NOT_FOUND) if row["provider_subscription_id"]: return self.send_json({"error": "A fizetési szolgáltatónál történő lemondás még nincs beállítva; az előfizetés ezért nem változott."}, HTTPStatus.SERVICE_UNAVAILABLE) db.execute("UPDATE subscriptions SET status='cancelled', cancelled_at=?, updated_at=? WHERE user_id=?", (timestamp, timestamp, user_id)) self.send_json({"ok": True, "status": "cancelled", "cancelledAt": timestamp, "message": "Az előfizetést lemondtad. További terhelés nem indul a GazdaTárs rendszeréből."}) def restore_backup(self, user_id, data): try: backup_id = int(data.get("backupId")) except (TypeError, ValueError): raise ValueError("Érvénytelen biztonsági mentés.") with connect() as db: row = db.execute("SELECT payload FROM backups WHERE id=? AND user_id=?", (backup_id, user_id)).fetchone() if not row: return self.send_json({"error": "A mentés nem található."}, HTTPStatus.NOT_FOUND) self.send_json({"ok": True, "payload": sanitize_backup_payload(json.loads(row[0]))}) def send_to_adviser(self, user_id, data): if data.get("confirmed") is not True: return self.send_json({"error": "A küldéshez felhasználói jóváhagyás szükséges."}, HTTPStatus.BAD_REQUEST) recipient = str(data.get("email", "")).strip() adviser_name = str(data.get("name", "")).strip()[:200] if not re.fullmatch(r"[^@\s]+@[^@\s]+\.[^@\s]+", recipient): raise ValueError("A falugazdász e-mail-címe érvénytelen.") filename = re.sub(r"[^\w.-]", "_", str(data.get("filename", "GazdaTars_eGN_2026.xlsx")), flags=re.UNICODE)[:160] try: backup_id = int(data.get("backupId")) if data.get("backupId") else None except (TypeError, ValueError): backup_id = None try: attachment = base64.b64decode(data.get("attachment", ""), validate=True) except (ValueError, TypeError): raise ValueError("Az Excel-csatolmány hibás.") if not attachment or len(attachment) > 4 * 1024 * 1024 or attachment[:2] != b"PK": raise ValueError("Az Excel-csatolmány hiányzik vagy túl nagy.") host = os.environ.get("SMTP_HOST") sender = os.environ.get("SMTP_FROM") or os.environ.get("SMTP_USER") if not host or not sender: self.log_delivery(user_id, adviser_name, recipient, filename, backup_id, "sikertelen", "Az e-mail-küldés nincs beállítva a szerveren.") return self.send_json({"error": "Az e-mail-küldés nincs beállítva a szerveren."}, HTTPStatus.SERVICE_UNAVAILABLE) message = EmailMessage() message["From"] = sender message["To"] = recipient message["Subject"] = "GazdaTárs – 2026-os eGN Excel" greeting = f"Tisztelt {adviser_name}!" if adviser_name else "Tisztelt Falugazdász!" message.set_content(f"{greeting}\n\nA GazdaTárs felhasználója ({user_id}) elküldte a kitöltött 2026-os eGN Excel-fájlt.\n\nÜdvözlettel:\nGazdaTárs") message.add_attachment(attachment, maintype="application", subtype="vnd.openxmlformats-officedocument.spreadsheetml.sheet", filename=filename) port = int(os.environ.get("SMTP_PORT", "465" if os.environ.get("SMTP_SSL") == "1" else "587")) smtp_class = smtplib.SMTP_SSL if os.environ.get("SMTP_SSL") == "1" else smtplib.SMTP try: with smtp_class(host, port, timeout=20) as smtp: if smtp_class is smtplib.SMTP and os.environ.get("SMTP_STARTTLS", "1") == "1": smtp.starttls() if os.environ.get("SMTP_USER"): smtp.login(os.environ["SMTP_USER"], os.environ.get("SMTP_PASSWORD", "")) smtp.send_message(message) except (OSError, smtplib.SMTPException) as exc: print(f"E-mail sending failed: {exc}") self.log_delivery(user_id, adviser_name, recipient, filename, backup_id, "sikertelen", "SMTP-küldési hiba") return self.send_json({"error": "Az e-mail küldése sikertelen. Próbáld újra később."}, HTTPStatus.BAD_GATEWAY) self.log_delivery(user_id, adviser_name, recipient, filename, backup_id, "sikeres", None) self.send_json({"ok": True, "message": f"Az Excel sikeresen elküldve: {recipient}"}) @staticmethod def log_delivery(user_id, name, email, filename, backup_id, status, error): timestamp = now_iso() with connect() as db: db.execute("INSERT OR IGNORE INTO users (user_id,created_at,updated_at) VALUES (?, ?, ?)", (user_id, timestamp, timestamp)) if backup_id and not db.execute("SELECT 1 FROM backups WHERE id=? AND user_id=?", (backup_id, user_id)).fetchone(): backup_id = None db.execute( "INSERT INTO email_deliveries(user_id, created_at, recipient_name, recipient_email, filename, backup_id, status, error_message) VALUES (?, ?, ?, ?, ?, ?, ?, ?)", (user_id, timestamp, name, email, filename, backup_id, status, error), ) if __name__ == "__main__": port = int(os.environ.get("PORT", "8000")) host = os.environ.get("HOST", "127.0.0.1") connect().close() print(f"GazdaTárs: http://localhost:{port}") ThreadingHTTPServer((host, port), Handler).serve_forever()