From b93e6ce0d6fa93bab70cc39407b6e2f19c94244b Mon Sep 17 00:00:00 2001 From: Luna Date: Mon, 21 Jul 2025 18:59:54 -0300 Subject: [PATCH] initial commit --- .gitignore | 2 + mullvad_check.py | 234 +++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 236 insertions(+) create mode 100644 .gitignore create mode 100644 mullvad_check.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..2a710aa --- /dev/null +++ b/.gitignore @@ -0,0 +1,2 @@ +.envrc +mullvad_cache.json diff --git a/mullvad_check.py b/mullvad_check.py new file mode 100644 index 0000000..9d7f8ca --- /dev/null +++ b/mullvad_check.py @@ -0,0 +1,234 @@ +#!/usr/bin/env python3 +import requests +from bs4 import BeautifulSoup +import json +import os +import time +from datetime import datetime, timezone + + +CACHE_FILE = "mullvad_cache.json" +CACHE_EXPIRY_DAYS = 10 +WARNING_DAYS = int( + os.getenv("MULLVAD_WARNING_DAYS", "5") +) # Configurable warning period +DISCORD_WEBHOOK_URL = os.getenv("DISCORD_WEBHOOK_URL") + + +def load_cache(): + """Load cache data from file if it exists and is recent""" + try: + with open(CACHE_FILE, "r") as f: + cache = json.load(f) + + # Check if cache is still valid (less than 10 days old) + time_diff = time.time() - cache.get("last_scrape", 0) + if time_diff < CACHE_EXPIRY_DAYS * 24 * 3600: # 10 days in seconds + print( + f"Using cached data from {datetime.fromtimestamp(cache['last_scrape'])}" + ) + return cache + except (FileNotFoundError, json.JSONDecodeError, KeyError): + pass + + return None + + +def save_cache( + expiry_datetime, + last_scrape, + last_notification_sent=None, + notification_sent_for_expiry=None, +): + """Save cache data to file""" + cache_data = { + "account_expiry": expiry_datetime, + "last_scrape": last_scrape, + "last_notification_sent": last_notification_sent, + "notification_sent_for_expiry": notification_sent_for_expiry, + } + with open(CACHE_FILE, "w") as f: + json.dump(cache_data, f, indent=2) + print(f"Cache saved to {CACHE_FILE}") + + +def send_discord_notification(expiry_datetime, days_remaining): + """Send Discord webhook notification""" + if not DISCORD_WEBHOOK_URL: + print("No Discord webhook URL configured, skipping notification") + return False + + print("sending Discord notification") + message = { + "content": f"⚠️ **Mullvad VPN Warning** ⚠️\n\nYour Mullvad account expires in **{days_remaining} days** on {expiry_datetime}!" + } + + try: + response = requests.post(DISCORD_WEBHOOK_URL, json=message) + if response.ok: + print( + f"Discord notification sent successfully (expires in {days_remaining} days)" + ) + return True + else: + print(f"Failed to send Discord notification: {response.status_code}") + return False + except Exception as e: + print(f"Error sending Discord notification: {e}") + return False + + +def calculate_days_remaining(expiry_datetime_str): + """Calculate days remaining until expiry""" + try: + # Parse the datetime string (assuming ISO format) + expiry_dt = datetime.fromisoformat(expiry_datetime_str.replace("Z", "+00:00")) + now = datetime.now(timezone.utc) + time_diff = expiry_dt - now + return time_diff.days + except Exception as e: + print(f"Error parsing expiry date: {e}") + return None + + +def main(): + # Check cache first + cache = load_cache() + if cache: + expiry_datetime = cache["account_expiry"] + days_remaining = calculate_days_remaining(expiry_datetime) + print(f"Account expires: {expiry_datetime}") + if days_remaining is not None: + print(f"Days remaining: {days_remaining}") + + # Check if we should send Discord notification even with cached data + if days_remaining <= WARNING_DAYS: + # Check if we've already sent notification for this expiry date + if cache.get("notification_sent_for_expiry") != expiry_datetime: + current_time = time.time() + if send_discord_notification(expiry_datetime, days_remaining): + # Update cache with notification info + save_cache( + expiry_datetime, + cache["last_scrape"], + current_time, + expiry_datetime, + ) + return + + session = requests.Session() + + # Step 1: POST login request + login_url = "https://mullvad.net/en/account/login" + + # Get account number from environment variable or prompt + account_number = os.getenv("MULLVAD_ACCOUNT_NUMBER") + if not account_number: + account_number = input("Enter your Mullvad account number: ") + + login_data = {"account_number": account_number} + + headers = { + "User-Agent": "Mozilla/5.0 (X11; Linux x86_64; rv:138.0) Gecko/20100101 Firefox/138.0", + "Accept": "application/json", + "Accept-Language": "en-US,en;q=0.5", + "Accept-Encoding": "gzip, deflate, br, zstd", + "Referer": "https://mullvad.net/en/account/login", + "Content-Type": "application/x-www-form-urlencoded", + "x-sveltekit-action": "true", + "Origin": "https://mullvad.net", + "Connection": "keep-alive", + "Sec-Fetch-Dest": "empty", + "Sec-Fetch-Mode": "cors", + "Sec-Fetch-Site": "same-origin", + "Priority": "u=0", + "Pragma": "no-cache", + "Cache-Control": "no-cache", + "TE": "trailers", + } + + print("Attempting login...") + login_response = session.post(login_url, data=login_data, headers=headers) + + print(f"Login response status: {login_response.status_code}") + + # Check if login was successful + if not login_response.ok: + print(f"Login failed with status {login_response.status_code}") + exit(1) + + response_json = login_response.json() + print(f"Login response JSON: {json.dumps(response_json, indent=2)}") + + # Print cookies to see if accessToken was set + print(f"Cookies after login: {dict(session.cookies)}") + + # Step 2: GET account page with cookie + account_url = "https://mullvad.net/en/account" + + print("\nFetching account page...") + account_response = session.get(account_url) + + print(f"Account page status: {account_response.status_code}") + + # Parse HTML with BeautifulSoup + soup = BeautifulSoup(account_response.text, "html.parser") + + print(f"Page title: {soup.title.string if soup.title else 'No title'}") + + expiry_element = soup.find("time", {"data-cy": "account-expiry"}) + assert expiry_element, "expiry element not found" + expiry_datetime = expiry_element["datetime"] + print(f"Account expires: {expiry_datetime}") + + # Calculate days remaining + days_remaining = calculate_days_remaining(expiry_datetime) + if days_remaining is not None: + print(f"Days remaining: {days_remaining}") + + # Check if we should send Discord notification + should_notify = False + current_time = time.time() + + # Load existing cache to check notification state + existing_cache = None + try: + with open(CACHE_FILE, "r") as f: + existing_cache = json.load(f) + except (FileNotFoundError, json.JSONDecodeError): + pass + + if days_remaining <= WARNING_DAYS: + # Check if we've already sent notification for this expiry date + if ( + not existing_cache + or existing_cache.get("notification_sent_for_expiry") != expiry_datetime + ): + should_notify = True + + # Send notification if needed + notification_sent_time = None + notification_sent_for_expiry = None + if should_notify and send_discord_notification(expiry_datetime, days_remaining): + notification_sent_time = current_time + notification_sent_for_expiry = expiry_datetime + elif existing_cache: + # Preserve existing notification state + notification_sent_time = existing_cache.get("last_notification_sent") + notification_sent_for_expiry = existing_cache.get( + "notification_sent_for_expiry" + ) + + save_cache( + expiry_datetime, + current_time, + notification_sent_time, + notification_sent_for_expiry, + ) + else: + current_time = time.time() + save_cache(expiry_datetime, current_time) + + +if __name__ == "__main__": + main()