diff --git a/README.md b/README.md index 35dbd13..8183ba9 100644 --- a/README.md +++ b/README.md @@ -1,63 +1,57 @@ # Powerpal Auth Extractor -Because the powerpal mobile app is not open source, and does not provide a way to extract the API key and serial number from the device, this script was created to extract the API key and serial number from the device directly via Bluetooth Low Energy (BLE). +Extracts the **Device Serial** and **API Key** from a Powerpal energy monitor via Bluetooth, without needing the Powerpal app. Once you have these credentials you can query the Powerpal API directly, or use them with integrations like Home Assistant. -Once extracted, these credentials can be used to query your home's energy usage data straight from the official Powerpal API endpoints. Can be used with integrations like Home Assistant, Grafana, etc. +## Requirements -## What it does +- Python 3.10+ +- Bluetooth enabled on your computer +- Your Powerpal's **6-digit pairing code** (on the card that came with your Powerpal, or in the Powerpal app) -This script uses your computer's Bluetooth to connect to your Powerpal. It authenticates using your 6-digit pairing code, reads the specific GATT characteristics holding your Device Serial Number and API Key (UUID), and formats them for use. - -## Prerequisites - -- **Python 3.7+** -- A computer with **Bluetooth enabled** (Windows, macOS, or Linux) -- Your Powerpal's **MAC Address** (Format: `12:34:56:78:9A:BC`) -- Your Powerpal's **6-digit Pairing Code** - -> **Note:** The MAC address and pairing code can typically be found printed on the device itself, on the original packaging card, or mostly within the settings of **You can also find the MAC address within bluetooth settings on your phone or computer.** - -## Installation - -1. Clone or download this repository. -2. Install the required dependencies: +## Setup ```bash pip install -r requirements.txt ``` -## How to use it - -### IMPORTANT: When and How to Connect via Bluetooth - -**DO NOT** try to pair or connect to the Powerpal device through your computer's operating system Bluetooth menu. +## Usage -The Powerpal device can only maintain **one active Bluetooth connection at a time**. In order for this script to work, you must ensure the Powerpal is completely disconnected from any other devices. - -**Before running the script:** -1. Ensure your computer's Bluetooth hardware is **turned on**. -2. **Disconnect your phone** from the Powerpal device. The easiest way to do this is to fully close/kill the Powerpal mobile app on your smartphone, or temporarily turn off Bluetooth on your phone altogether. -3. The Python script will handle the Bluetooth connection directly. *make sure you are within range of the Powerpal device* - -### Running the script - -You can run the script and manually enter the information when prompted: +Close the Powerpal app on your phone (or disable phone Bluetooth) so the device is free to connect, then run: ```bash -python retrieve_api_key.py +$ python retrieve_api_key.py ``` -Or you can provide your MAC address and Pairing code directly as command-line arguments: +``` +Found: Powerpal 000123ab (12:34:56:78:9A:BC) +? Your Powerpal pairing code: 123456 +✓ Paired +╭─────────────────────────────────── Powerpal Credentials ───────────────────────────────────╮ +│ Serial (Device ID) 000123ab │ +│ API key (Authorisation Key) 7e4a9f12-83cb-4d56-ae01-2f9b0c3d7e58 │ +│ Readings endpoint https://readings.powerpal.net/api/v1/meter_reading/000123ab │ +╰────────────────────────────────────────────────────────────────────────────────────────────╯ +? Validate credentials by calling the Powerpal API for device 000123ab? Yes +✓ Credentials verified — your Device ID and Authorisation Key were successfully retrieved! +╭──────────────────────────────────────── API Response ──────────────────────────────────────╮ +│ { │ +│ "serial_number": "000123ab", │ +│ "total_meter_reading_count": 312847, │ +│ "total_watt_hours": 2914830, │ +│ "total_cost": 1053.42, │ +│ ... │ +│ } │ +╰────────────────────────────────────────────────────────────────────────────────────────────╯ +``` + +You can pass arguments directly to skip the prompts: ```bash -python retrieve_api_key.py 12:34:56:78:9A:BC 123456 +$ python retrieve_api_key.py --code 123456 +$ python retrieve_api_key.py --code 123456 --device 12:34:56:78:9A:BC ``` -### Script Execution - -1. The script will pause and ask you to confirm that no other devices (like your phone) are currently connected to the Powerpal. Hit `Enter` to proceed. -2. It attempts to connect to the Powerpal via BLE and authenticates with your pairing code. -3. It grabs your **Device Serial**. -4. It grabs your **API Key (UUID)**. -5. It will finally ask `(y/n)` if you'd like to test the credentials against `https://readings.powerpal.net/api/v1/device/{serial}`. -6. If the request succeeds, your Powerpal's configuration metadata will be printed to the console, and you now have the required keys to build your own integrations (like Home Assistant, Grafana, etc.). +| Option | Description | +| --- | --- | +| `--code` | 6-digit pairing code | +| `--device` | Bluetooth MAC address — skips scanning if you already know it | diff --git a/requirements.txt b/requirements.txt index b4069a4..9086ba0 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,2 +1,6 @@ bleak requests +rich +questionary +typer +dbus-fast; sys_platform == "linux" diff --git a/retrieve_api_key.py b/retrieve_api_key.py index 5f5a87e..39a13c2 100644 --- a/retrieve_api_key.py +++ b/retrieve_api_key.py @@ -1,74 +1,280 @@ -from encodings import utf_8 -import sys +from __future__ import annotations + import asyncio +import sys +from typing import Final + import requests +import typer +from bleak import BleakClient, BleakError, BleakScanner +from rich.console import Console +from rich.panel import Panel +from rich.table import Table +import questionary + +console = Console() + +# Powerpal GATT characteristic UUIDs +PAIRING_CODE_CHAR: Final[str] = '59DA0011-12F4-25A6-7D4F-55961DCE4205' +POWERPAL_UUID_CHAR: Final[str] = '59DA0009-12F4-25A6-7D4F-55961DCE4205' +POWERPAL_SERIAL_CHAR: Final[str] = '59DA0010-12F4-25A6-7D4F-55961DCE4205' + +MAX_RETRIES: Final[int] = 3 + +if sys.platform == 'linux': + from dbus_fast.aio import MessageBus + from dbus_fast.service import ServiceInterface, method + from dbus_fast import BusType + + AGENT_PATH = '/com/powerpalextractor/Agent' + + class PairingAgent(ServiceInterface): + """BlueZ pairing agent for Just Works bonding without user prompts. + + AuthorizeService and RequestAuthorization must return success or BlueZ + will abort the connection after bonding. + """ + + def __init__(self) -> None: + super().__init__('org.bluez.Agent1') + + @method() + def Release(self): + pass -from bleak import BleakClient, BleakScanner, BleakError - -pairingCodeChar = '59DA0011-12F4-25A6-7D4F-55961DCE4205' -powerpalUUIDChar ='59DA0009-12F4-25A6-7D4F-55961DCE4205' -powerpalSerialChar = '59DA0010-12F4-25A6-7D4F-55961DCE4205' - -def convert_pairing_code(original_pairing_code): - return int(original_pairing_code).to_bytes(4, byteorder='little') - -async def main(address, my_pairing_code): - while address is None: - address = input("Your Powerpal MAC address: ") - if (address.count(':') != 5) or (len(address) != 17): - address = None - print("Incorrect MAC address formatting, should look like -> 12:34:56:78:9A:BC") - while my_pairing_code is None: - my_pairing_code = int(input("Your Powerpal pairing code: ")) - if not (0 <= my_pairing_code <= 999999): - my_pairing_code = None - print("Pairing Code should be 6 digits...") - input("Please confirm that you are NOT connected to the Powerpal via Bluetooth using any devices, and that bluetooth is enabled on your computer, and hit enter to continue...") - - async with BleakClient(address) as client: - print(f"Connected: {client.is_connected}") - # Some Bleak backends (e.g. Windows WinRT) do not accept a pairing kind argument. - # Call pair() without arguments for maximum compatibility. + @method() + def RequestAuthorization(self, device: 'o'): # type: ignore[override] + pass + + @method() + def AuthorizeService(self, device: 'o', uuid: 's'): # type: ignore[override] + pass + + @method() + def Cancel(self): + pass + + + async def register_agent(bus: 'MessageBus') -> 'PairingAgent': + agent = PairingAgent() + bus.export(AGENT_PATH, agent) + introspection = await bus.introspect('org.bluez', '/org/bluez') + bluez_proxy = bus.get_proxy_object('org.bluez', '/org/bluez', introspection) + agent_mgr = bluez_proxy.get_interface('org.bluez.AgentManager1') + await agent_mgr.call_register_agent(AGENT_PATH, 'NoInputNoOutput') + await agent_mgr.call_request_default_agent(AGENT_PATH) + return agent + + + async def unregister_agent(bus: 'MessageBus') -> None: try: - paired = await client.pair() - print(f"Paired?: {paired}") - except TypeError: - # Older/newer Bleak versions may not support pair() on this backend at all. - # In that case, continue without explicit pairing and rely on connect(). - print("Pairing method not supported on this backend; continuing without explicit pair().") - - print(f"Authenticating with pairing_code: {my_pairing_code}, converted: {convert_pairing_code(my_pairing_code)}") - await client.write_gatt_char(pairingCodeChar, convert_pairing_code(my_pairing_code), response=False) - print('Auth Success\n') - - await asyncio.sleep(0.5) - - print(f"Attempting to retrieve Powerpal Serial Number...") - serial = await client.read_gatt_char(powerpalSerialChar) - formatted_serial = ''.join('{:02x}'.format(x) for x in reversed(serial)).lower() - print(f"Retrieved Device Serial: {formatted_serial}") - print(f"Meaning your meter reading endpoint is: https://readings.powerpal.net/api/v1/meter_reading/{formatted_serial}\n") - - print(f"Attempting to retrieve apikey (Powerpal UUID)...") - apikey = await client.read_gatt_char(powerpalUUIDChar) - joined_apikey = ''.join('{:02x}'.format(x) for x in apikey) - formatted_apikey = f"{joined_apikey[:8]}-{joined_apikey[8:12]}-{joined_apikey[12:16]}-{joined_apikey[16:20]}-{joined_apikey[20:]}".lower() - print(f"Retrieved apikey: {formatted_apikey}\n") - - test_endpoint = input(f"Would you like this script to validate your apikey by using it to make a request for your device information from https://readings.powerpal.net/api/v1/device/{formatted_serial}? (y/n): ") - if test_endpoint.lower().startswith('y'): - url = f"https://readings.powerpal.net/api/v1/device/{formatted_serial}" - headers = { 'Authorization': formatted_apikey } - - response = requests.request("GET", url, headers=headers) - if (response.status_code < 300): - print("Success! Looks like your Device Serial and apikey were succesfully retrieved! Here is the device information returned from https://readings.powerpal.net/api/ :") - print(response.text) - else: - print("Failure!") - print(f"{response.status_code} - {response.reason} - {response.text}") + introspection = await bus.introspect('org.bluez', '/org/bluez') + bluez_proxy = bus.get_proxy_object('org.bluez', '/org/bluez', introspection) + agent_mgr = bluez_proxy.get_interface('org.bluez.AgentManager1') + await agent_mgr.call_unregister_agent(AGENT_PATH) + except Exception: + pass + +def convert_pairing_code(pairing_code: int) -> bytes: + return pairing_code.to_bytes(4, byteorder='little') + + +async def scan_for_powerpal() -> str: + """Scan repeatedly until a Powerpal is found, then return its address.""" + while True: + with console.status("Scanning for Powerpal devices..."): + powerpals = [d for d in await BleakScanner.discover(timeout=5.0) + if d.name and d.name.startswith('Powerpal')] + if len(powerpals) == 1: + console.print(f"[green]Found:[/green] {powerpals[0].name} ({powerpals[0].address})") + return powerpals[0].address + if len(powerpals) > 1: + choice = await questionary.select( + "Multiple Powerpal devices found. Select one:", + choices=[f"{d.name} ({d.address})" for d in powerpals], + ).ask_async() + if choice: + return next(d.address for d in powerpals if f"{d.name} ({d.address})" == choice) + + +async def get_pairing_code() -> int | None: + """Prompt the user for a valid 6-digit pairing code. Returns None if cancelled.""" + raw = await questionary.text( + "Your Powerpal pairing code:", + validate=lambda v: True if (v.isdigit() and 0 <= int(v) <= 999999) else "Must be a 6-digit number", + ).ask_async() + if raw is None: + return None + return int(raw) + + +async def pair_with_device(device: str, pairing_code: int) -> bool | None: + """Attempt to pair with the Powerpal. + + Returns: + True – paired successfully. + False – pairing code was wrong; caller should re-prompt. + None – user aborted the retry prompt; caller should exit. + Raises BleakError after exhausting MAX_RETRIES connection attempts. + """ + for attempt in range(1, MAX_RETRIES + 1): + try: + with console.status(f"Pairing with {device}..."): + async with BleakClient(device, timeout=30.0) as client: + await client.write_gatt_char(PAIRING_CODE_CHAR, convert_pairing_code(pairing_code), response=False) + await client.read_gatt_char(POWERPAL_SERIAL_CHAR) # validate pairing code is correct + console.print("[green]\u2713[/green] Paired") + return True + except BleakError as e: + if 'NotPermitted' in str(e): + console.print("[red]\u2717[/red] Incorrect pairing code.") + return False + console.print(f"[yellow]Pairing attempt {attempt} failed:[/yellow] {e}") + if attempt < MAX_RETRIES: + ok = await questionary.confirm( + "Ensure Bluetooth is enabled and no other device is connected to the Powerpal. Ready to retry?", + default=True, + ).ask_async() + if not ok: + return None # user chose to abort + else: + raise + return None # unreachable + + +def format_serial(data: bytes) -> str: + """Convert 4-byte little-endian serial to a lowercase hex string.""" + if len(data) != 4: + raise ValueError(f"Unexpected serial length: {len(data)} bytes (expected 4)") + if not any(data): + raise ValueError("Serial is all zeros") + return ''.join('{:02x}'.format(x) for x in reversed(data)) + + +def format_apikey(data: bytes) -> str: + """Convert 16-byte API key to lowercase UUID format string.""" + if len(data) != 16: + raise ValueError(f"Unexpected apikey length: {len(data)} bytes (expected 16)") + if not any(data): + raise ValueError("API key is all zeros") + h = ''.join('{:02x}'.format(x) for x in data) + return f"{h[:8]}-{h[8:12]}-{h[12:16]}-{h[16:20]}-{h[20:]}" + + +async def read_credentials(device: str, pairing_code: int) -> tuple[str, str]: + """Authenticate and read (serial, apikey) from the device, retrying up to MAX_RETRIES times.""" + for attempt in range(1, MAX_RETRIES + 1): + try: + with console.status("Authenticating and reading credentials..."): + async with BleakClient(device, timeout=30.0) as client: + await client.write_gatt_char(PAIRING_CODE_CHAR, convert_pairing_code(pairing_code), response=False) + serial_bytes = await client.read_gatt_char(POWERPAL_SERIAL_CHAR) + apikey_bytes = await client.read_gatt_char(POWERPAL_UUID_CHAR) + return format_serial(serial_bytes), format_apikey(apikey_bytes) + except (BleakError, ValueError) as e: + console.print(f"[yellow]Read attempt {attempt} failed:[/yellow] {e}") + if attempt < MAX_RETRIES: + await asyncio.sleep(3) + else: + raise + raise RuntimeError("unreachable") # satisfies type checker + + +def display_credentials(serial: str, apikey: str) -> None: + """Render the credentials panel to the console.""" + table = Table(show_header=False, box=None, padding=(0, 1)) + table.add_row("[bold]Serial[/bold] [dim](Device ID)[/dim]", serial) + table.add_row("[bold]API key[/bold] [dim](Authorisation Key)[/dim]", apikey) + table.add_row("[bold]Readings endpoint[/bold]", f"https://readings.powerpal.net/api/v1/meter_reading/{serial}") + console.print(Panel(table, title="[bold green]Powerpal Credentials[/bold green]", expand=False)) + + +async def validate_credentials(serial: str, apikey: str) -> None: + """Optionally validate credentials against the Powerpal API.""" + confirmed = await questionary.confirm( + f"Validate credentials by calling the Powerpal API for device {serial}?", + default=False, + ).ask_async() + if confirmed: + response = requests.get( + f"https://readings.powerpal.net/api/v1/device/{serial}", + headers={'Authorization': apikey}, + ) + if response.status_code < 300: + console.print("[green]\u2713[/green] Credentials verified \u2014 your Device ID and Authorisation Key were successfully retrieved!") + console.print(Panel(response.text, title="[bold green]API Response[/bold green]")) else: - print(f'You may also confirm your apikey by using curl:\ncurl -H "Authorization: {formatted_apikey}" https://readings.powerpal.net/api/v1/device/{formatted_serial}') + console.print(f"[red]Request failed:[/red] {response.status_code} {response.reason}\n{response.text}") + else: + console.print(f'[dim]Tip:[/dim] curl -H "Authorization: {apikey}" https://readings.powerpal.net/api/v1/device/{serial}') + + +async def main(device: str | None, pairing_code: int | None) -> None: + # Resolve device address + if device is None: + device = await scan_for_powerpal() + elif (device.count(':') != 5) or (len(device) != 17): + console.print("[red]Incorrect MAC address format.[/red] Should look like 12:34:56:78:9A:BC") + device = await scan_for_powerpal() + + # Resolve pairing code + if pairing_code is None: + pairing_code = await get_pairing_code() + if pairing_code is None: + return + + # On Linux, register a BlueZ D-Bus pairing agent so that the passkey + # exchange during BLE bonding completes automatically without user prompts. + # On macOS/Windows the OS handles pairing through its own native dialogs. + bus = None + if sys.platform == 'linux': + bus = await MessageBus(bus_type=BusType.SYSTEM).connect() + await register_agent(bus) + + try: + # Pairing loop — re-prompts for pairing code on wrong-code errors + while True: + result = await pair_with_device(device, pairing_code) + if result is True: + break + if result is None: + return # user aborted retry + # result is False: wrong pairing code — ask again + pairing_code = await get_pairing_code() + if pairing_code is None: + return + + with console.status("Waiting for device to settle..."): + await asyncio.sleep(2) + + serial, apikey = await read_credentials(device, pairing_code) + display_credentials(serial, apikey) + await validate_credentials(serial, apikey) + finally: + if bus is not None: + await unregister_agent(bus) + bus.disconnect() + + +app = typer.Typer(rich_markup_mode='rich') + + +@app.command() +def cli( + code: int | None = typer.Option( + default=None, + metavar="CODE", + help="6-digit pairing code from the card included with your Powerpal or the Powerpal app. If omitted, you will be prompted.", + ), + device: str | None = typer.Option( + default=None, + help="Bluetooth MAC address of the Powerpal (e.g. [bold]12:34:56:78:9A:BC[/bold]). " + "If omitted, the device will be located via scan.", + ), +) -> None: + """Extract the API key and serial number from a Powerpal energy monitor via Bluetooth.""" + asyncio.run(main(device, code)) + if __name__ == "__main__": - asyncio.run(main((sys.argv[1] if len(sys.argv) >= 2 else None),(sys.argv[2] if len(sys.argv) == 3 else None))) \ No newline at end of file + app() \ No newline at end of file