234 lines
7.9 KiB
Python
234 lines
7.9 KiB
Python
#!/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()
|