Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 6 additions & 9 deletions .github/workflows/merge.yml
Original file line number Diff line number Diff line change
Expand Up @@ -28,18 +28,15 @@ jobs:
uses: actions/setup-python@v5
with:
python-version: ${{ env.DEFAULT_PYTHON }}
- name: Prepare poetry
- name: Prepare uv
run: |
pip install uv
uv venv --seed venv
. venv/bin/activate
uv pip install poetry
- name: Dependencies and build
- name: Build
run: |
. venv/bin/activate
poetry install --no-root
poetry build
uv build
- name: Publish distribution 📦 to PyPI
uses: pypa/gh-action-pypi-publish@release/v1
with:
skip-existing: true
run: |
. venv/bin/activate
uv publish
17 changes: 6 additions & 11 deletions .github/workflows/verify.yml
Original file line number Diff line number Diff line change
Expand Up @@ -33,20 +33,15 @@ jobs:
steps:
- name: Check out committed code
uses: actions/checkout@v4
- name: Prepare poetry
- name: Prepare uv
run: |
pip install uv
uv venv --seed venv
- name: Build
run: |
. venv/bin/activate
uv pip install poetry
- name: Dependencies and build
uv build
- name: Publish distribution 📦 to TestPyPI
run: |
. venv/bin/activate
poetry install --no-root
poetry build
- name: Publish distribution 📦 to Test PyPI
uses: pypa/gh-action-pypi-publish@release/v1
continue-on-error: true
with:
repository-url: https://test.pypi.org/legacy/
skip-existing: true
uv publish --publish-url https://test.pypi.org/legacy/
120 changes: 69 additions & 51 deletions airos/airos8.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,12 +2,10 @@

from __future__ import annotations

from typing import cast

import asyncio
import json
import logging
import ssl
import ssl

import aiohttp

Expand All @@ -19,14 +17,15 @@
class AirOS8:
"""Set up connection to AirOS."""

def __init__(self, host: str, username: str, password: str): -> None
def __init__(self, host: str, username: str, password: str):
"""Initialize AirOS8 class."""
self.username = username
self.password = password
self.base_url = f"https://{host}"

self._login_url = f"{self.base_url}/api/auth" # AirOS 8
self._status_cgi_url = f"{self.base_url}/status.cgi" # AirOS 8
self.current_csrf_token = None

self._use_json_for_login_post = False

Expand All @@ -44,20 +43,15 @@ def __init__(self, host: str, username: str, password: str): -> None
"X-Requested-With": "XMLHttpRequest",
}

async def login(self): -> bool
async def login(self) -> bool:
"""Log in to the device assuring cookies and tokens set correctly."""
loop = asyncio.get_running_loop()
ssl_context = await loop.run_in_executor(
None,
ssl.create_default_context
)
ssl_context = await loop.run_in_executor(None, ssl.create_default_context)
ssl_context.check_hostname = False
ssl_context.verify_mode = ssl.CERT_NONE
connector = aiohttp.TCPConnector(ssl=ssl_context)

async with aiohttp.ClientSession(connector=connector) as self.session:
current_csrf_token = None

# --- Step 0: Pre-inject the 'ok=1' cookie before login POST (mimics curl) ---
self.session.cookie_jar.update_cookies({"ok": "1"})

Expand All @@ -74,26 +68,43 @@ async def login(self): -> bool
login_request_headers["Content-Type"] = "application/json"
post_data = json.dumps(login_payload)
else:
login_request_headers["Content-Type"] = "application/x-www-form-urlencoded; charset=UTF-8"
login_request_headers["Content-Type"] = (
"application/x-www-form-urlencoded; charset=UTF-8"
)
post_data = login_payload

try:
async with self.session.post(self._login_url, data=post_data, headers=login_request_headers) as response:
async with self.session.post(
self._login_url, data=post_data, headers=login_request_headers
) as response:
if not response.cookies:
logger.exception("Empty cookies after login, bailing out.")
raise DataMissingError
raise DataMissingError from None
else:
for _, morsel in response.cookies.items():

# If the AIROS_ cookie was parsed but isn't automatically added to the jar, add it manually
if morsel.key.startswith("AIROS_") and morsel.key not in self.session.cookie_jar:
if (
morsel.key.startswith("AIROS_")
and morsel.key not in self.session.cookie_jar
):
# `SimpleCookie`'s Morsel objects are designed to be compatible with cookie jars.
# We need to set the domain if it's missing, otherwise the cookie might not be sent.
# For IP addresses, the domain is typically blank.
# aiohttp's jar should handle it, but for explicit control:
if not morsel.get("domain"):
morsel["domain"] = response.url.host # Set to the host that issued it
self.session.cookie_jar.update_cookies({morsel.key: morsel.output(header="")[len(morsel.key)+1:].split(";")[0].strip()}, response.url)
morsel["domain"] = (
response.url.host
) # Set to the host that issued it
self.session.cookie_jar.update_cookies(
{
morsel.key: morsel.output(header="")[
len(morsel.key) + 1 :
]
.split(";")[0]
.strip()
},
response.url,
)
# The update_cookies method can take a SimpleCookie morsel directly or a dict.
# The morsel.output method gives 'NAME=VALUE; Path=...; HttpOnly'
# We just need 'NAME=VALUE' or the morsel object itself.
Expand All @@ -104,68 +115,75 @@ async def login(self): -> bool
# Simpler: just ensure the key-value pair is there for simple jar.

# Let's try the direct update of the key-value
self.session.cookie_jar.update_cookies({morsel.key: morsel.value})
self.session.cookie_jar.update_cookies(
{morsel.key: morsel.value}
)

new_csrf_token = response.headers.get("X-CSRF-ID")
if new_csrf_token:
current_csrf_token = new_csrf_token
self.current_csrf_token = new_csrf_token
else:
return

# Re-check cookies in self.session.cookie_jar AFTER potential manual injection
airos_cookie_found = False
ok_cookie_found = False
if not self.session.cookie_jar:
logger.exception("COOKIE JAR IS EMPTY after login POST. This is a major issue.")
raise DataMissingError
logger.exception(
"COOKIE JAR IS EMPTY after login POST. This is a major issue."
)
raise DataMissingError from None
for cookie in self.session.cookie_jar:
if cookie.key.startswith("AIROS_"):
airos_cookie_found = True
if cookie.key == "ok":
ok_cookie_found = True

if not airos_cookie_found and not ok_cookie_found:
raise DataMissingError
raise DataMissingError from None

response_text = await response.text()

if response.status == 200:
try:
json.loads(response_text)
return True
except json.JSONDecodeError:
except json.JSONDecodeError as err:
logger.exception("JSON Decode Error")
raise DataMissingError
raise DataMissingError from err

else:
log = f"Login failed with status {response.status}. Full Response: {response.text}"
logger.error(log)
raise ConnectionFailedError
except aiohttp.ClientError:
raise ConnectionFailedError from None
except aiohttp.ClientError as err:
logger.exception("Error during login")
raise ConnectionFailedError
raise ConnectionFailedError from err

async def status(self): -> dict
async def status(self) -> dict:
"""Retrieve status from the device."""
# --- Step 2: Verify authenticated access by fetching status.cgi ---
authenticated_get_headers = {**self._common_headers}
if current_csrf_token:
authenticated_get_headers["X-CSRF-ID"] = current_csrf_token

try:
async with self.session.get(self._status_cgi_url, headers=authenticated_get_headers) as response:
status_response_text = await response.text()

if response.status == 200:
try:
return json.loads(status_response_text)
except json.JSONDecodeError:
logger.exception("JSON Decode Error in authenticated status response")
raise DataMissingError
else:
log = f"Authenticated status.cgi failed: {response.status}. Response: {status_response_text}"
logger.error(log)
except aiohttp.ClientError:
logger.exception("Error during authenticated status.cgi call")
raise ConnectionFailedError

# --- Step 2: Verify authenticated access by fetching status.cgi ---
authenticated_get_headers = {**self._common_headers}
if self.current_csrf_token:
authenticated_get_headers["X-CSRF-ID"] = self.current_csrf_token

try:
async with self.session.get(
self._status_cgi_url, headers=authenticated_get_headers
) as response:
status_response_text = await response.text()

if response.status == 200:
try:
return json.loads(status_response_text)
except json.JSONDecodeError:
logger.exception(
"JSON Decode Error in authenticated status response"
)
raise DataMissingError from None
else:
log = f"Authenticated status.cgi failed: {response.status}. Response: {status_response_text}"
logger.error(log)
except aiohttp.ClientError as err:
logger.exception("Error during authenticated status.cgi call")
raise ConnectionFailedError from err
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"

[project]
name = "airos"
version = "0.0.3a0"
version = "0.0.3a1"
license = "MIT"
description = "Ubiquity airOS module(s) for Python 3."
readme = "README.md"
Expand Down
3 changes: 3 additions & 0 deletions requirements-test.txt
Original file line number Diff line number Diff line change
@@ -0,0 +1,3 @@
pytest
pytest-asyncio
aiohttp