Add sensor to Fritz!box integration for status of External acces. (wireguard VPN)

Hi, Fritz!box has a WireGuard feature for a Tunnel to your home network. When a device connects by using the WireGuard app on smartphones or an other device you can see it in the Fritz!box. External Access turns on.

Can we get a sensor in HA for this? I’d like to create a notification automation to keep track of the status and be notified when the tunnel is active.

Hi,

i think the AVM FRITZ!Box Tools Integration does not provide these details. However you could use AppDaemon or Pyscript to update a sensor.

Please be aware that the login info for your FritzBox is needed for that, which might introduce a security risk. I would definitely recommend to create a separate user inside your fritzbox (System > Fritz!Box Users) for that use case.

The following script could be a starting point, however it currently lacks the final implementation to create/update a HA sensor.

#!/usr/bin/env python3
"""
FRITZ!OS WebGUI Login
Get a sid (session ID) via PBKDF2 based challenge response algorithm.
Fallback to MD5 if FRITZ!OS has no PBKDF2 support.
AVM 2020-09-25
edited on 2024-10-14 to obtain vpn info and send it to HA
"""
import sys
import json
import hashlib
import time
import urllib.request
import urllib.parse
import xml.etree.ElementTree as ET

LOGIN_SID_ROUTE = "/login_sid.lua?version=2"

def get_fritzbox_data(url:str, sid:str) -> dict[str, any]:
    # Define the URL
    url = f'{url}/data.lua'

    # Define the headers
    headers = {
        'Accept': '*/*',
        'Accept-Language': 'de-DE,de;q=0.9,en-US;q=0.8,en;q=0.7',
        'Connection': 'keep-alive',
        'Content-Type': 'application/x-www-form-urlencoded',
        'Origin': f'http://{url}',
        'Referer': f'http://{url}/',
        'User-Agent': 'Mozilla/5.0 (iPhone; CPU iPhone OS 16_6 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/16.6 Mobile/15E148 Safari/604.1'
    }

    # Define the data
    data = {
        'xhr': '1',
        'sid': sid,
        'lang': 'de',
        'page': 'overview',
        'xhrId': 'all',
        'useajax': '1',
        'no_sidrenew': ''
    }
    data_encoded = urllib.parse.urlencode(data).encode('utf-8')
    req = urllib.request.Request(url, data=data_encoded, headers=headers)
    response = urllib.request.urlopen(req)
    response_data = response.read().decode('utf-8')
    return(json.loads(response_data))

class LoginState:
    def __init__(self, challenge: str, blocktime: int):
        self.challenge = challenge
        self.blocktime = blocktime
        self.is_pbkdf2 = challenge.startswith("2$")

def get_sid(box_url: str, username: str, password: str) -> str:
    """ Get a sid by solving the PBKDF2 (or MD5) challenge-response
    process.
    """
    try:
        state = get_login_state(box_url)
    except Exception as ex:
        raise Exception("failed to get challenge") from ex
    if state.is_pbkdf2:
        print("PBKDF2 supported")
        challenge_response = calculate_pbkdf2_response(state.challenge, password)
    else:
        print("Falling back to MD5")
        challenge_response = calculate_md5_response(state.challenge, password)
    if state.blocktime > 0:
        print(f"Waiting for {state.blocktime} seconds...")
        time.sleep(state.blocktime)
    try:
        sid = send_response(box_url, username, challenge_response)
    except Exception as ex:
        raise Exception("failed to login") from ex
    if sid == "0000000000000000":
        raise Exception("wrong username or password")
    return sid


def get_login_state(box_url: str) -> LoginState:
    """ Get login state from FRITZ!Box using login_sid.lua?version=2 """
    url = box_url + LOGIN_SID_ROUTE
    http_response = urllib.request.urlopen(url)
    xml = ET.fromstring(http_response.read())
    # print(f"xml: {xml}")
    challenge = xml.find("Challenge").text
    blocktime = int(xml.find("BlockTime").text)
    return LoginState(challenge, blocktime)

def calculate_pbkdf2_response(challenge: str, password: str) -> str:
    """ Calculate the response for a given challenge via PBKDF2 """
    challenge_parts = challenge.split("$")
    # Extract all necessary values encoded into the challenge
    iter1 = int(challenge_parts[1])
    salt1 = bytes.fromhex(challenge_parts[2])
    iter2 = int(challenge_parts[3])
    salt2 = bytes.fromhex(challenge_parts[4])
    # Hash twice, once with static salt...
    hash1 = hashlib.pbkdf2_hmac("sha256", password.encode(), salt1, iter1)
    # Once with dynamic salt.
    hash2 = hashlib.pbkdf2_hmac("sha256", hash1, salt2, iter2)
    return f"{challenge_parts[4]}${hash2.hex()}"

def calculate_md5_response(challenge: str, password: str) -> str:
    """ Calculate the response for a challenge using legacy MD5 """
    response = challenge + "-" + password
    # the legacy response needs utf_16_le encoding
    response = response.encode("utf_16_le")
    md5_sum = hashlib.md5()
    md5_sum.update(response)
    response = challenge + "-" + md5_sum.hexdigest()
    return response

def send_response(box_url: str, username: str, challenge_response: str) -> str:
    """ Send the response and return the parsed sid. raises an Exception on error """
    # Build response params
    post_data_dict = {"username": username, "response": challenge_response}
    post_data = urllib.parse.urlencode(post_data_dict).encode()
    headers = {"Content-Type": "application/x-www-form-urlencoded"}
    url = box_url + LOGIN_SID_ROUTE
    # Send response
    http_request = urllib.request.Request(url, post_data, headers)
    http_response = urllib.request.urlopen(http_request)
    # Parse SID from resulting XML.
    xml = ET.fromstring(http_response.read())
    return xml.find("SID").text


def logout(box_url: str, sid: str) -> bool:
    """ Send the logout requests to deactiate the session-id immediatly """
    # Build response params
    post_data_dict = {"logout": "", "sid": sid}
    post_data = urllib.parse.urlencode(post_data_dict).encode()
    headers = {"Content-Type": "application/x-www-form-urlencoded"}
    url = box_url + LOGIN_SID_ROUTE
    # Send response
    http_request = urllib.request.Request(url, post_data, headers)
    http_response = urllib.request.urlopen(http_request)
    # Parse SID from resulting XML.
    xml = ET.fromstring(http_response.read())
    return xml.find("SID").text == '0000000000000000' # will return True if logged out successfully

def main():
    url = "http://fritz.box" # the url of your router, can be an ip-address as well
    username = "MYUSERNAME" # the username; in the FritzBox Admin Panel go to: "System" > "Fritz!Box-Users". The username usually starts with "fritz..."
    password = "MYPASSWORD" # the password to log into the admin console; usually written somewhere on the back of your fritzbox

    # 1) Log into the fritzbox
    sid = get_sid(url, username, password)
    print(f"Successful login for user: {username}")

    # 2) Access all your current data as json
    data = get_fritzbox_data(url=url, sid=sid)

    # 3) Parse the relevant vpn information
    vpn_status = {vpn['name']:vpn['led'] for vpn in data['data']['vpn']['elements']} 
    # vpn_status is a dict with all vpn connections, e.g.: {"my_wireguard_connection1": "1", "my_wireguard_connection2": "0" , ...}
    # print all vpn connection and their status
    for k,v in vpn_status.items():
        print(f"VPN Connection with name {k} is {'being used' if(v=='1') else 'not being used'} at the moment.")

    # 4) Logout / deactive the session-id
    logout_state = logout(box_url=url, sid=sid)
    if(logout_state):
        print("You are logged out now!")

    # 5) Set the current vpn status
    for k,v in vpn_status.items():
        print(f"VPN Connection with name {k} is {'being used' if(v=='1') else 'not being used'} at the moment.")
        #### TODO ###
        #### set the sensor value using AppDaemon here ###
        
if __name__ == "__main__":
    main()

1 Like

Hi @Zwijgplicht
could you please provide some screenshots of the information you want to see in HA? A first search about vpns through the available apis only points to the X_AVM-DE_AppSetup service api. But maybe your remote device will be recognized by the mesh topology data which are already used inside the AVM Fritz!Tools integration :thinking:

Hi,

i actually figured out a much simpler way than my first described one (no coding needed):

  1. Find the IP-Address of your Wireguard VPN Connection → see the network tab of your fritzbox router (each Wireguard connection should point to a distinct IP-Adress outside of your DHCP range, e.g. xxx.xxx.xxx.200)

  2. Install / Activate the “Ping” Integration of homeassistant (see here)

  3. Add a new device using the installed ping integrations (Go to Settings → Integrations → Ping: add new device. Use the IP-Address from Step 1)

  4. Finished. You can now use the created device as a device tracker