initial commit
This commit is contained in:
commit
b93e6ce0d6
2 changed files with 236 additions and 0 deletions
234
mullvad_check.py
Normal file
234
mullvad_check.py
Normal file
|
|
@ -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()
|
||||
Loading…
Add table
Add a link
Reference in a new issue