sunnylink: centralize key pair handling in sunnylink registration (#1510)

* refactor(api): centralize key pair handling in SunnyLink registration

- Replaced manual key reading with `BaseApi.get_key_pair` for consistency.
- Simplifies code and improves maintainability of key management.

* cleanup
This commit is contained in:
DevTekVE
2025-11-24 23:33:40 +01:00
committed by GitHub
parent 844f4cbc74
commit c53e2134e2
+3 -8
View File
@@ -3,7 +3,6 @@ import os
import random import random
import time import time
from datetime import datetime, timedelta from datetime import datetime, timedelta
from pathlib import Path
import jwt import jwt
from openpilot.common.api.base import BaseApi from openpilot.common.api.base import BaseApi
@@ -81,23 +80,19 @@ class SunnylinkApi(BaseApi):
if sunnylink_dongle_id not in (None, UNREGISTERED_SUNNYLINK_DONGLE_ID): if sunnylink_dongle_id not in (None, UNREGISTERED_SUNNYLINK_DONGLE_ID):
return sunnylink_dongle_id return sunnylink_dongle_id
privkey_path = Path(f"{Paths.persist_root()}/comma/id_rsa") jwt_algo, private_key, public_key = BaseApi.get_key_pair()
pubkey_path = Path(f"{Paths.persist_root()}/comma/id_rsa.pub")
start_time = time.monotonic() start_time = time.monotonic()
successful_registration = False successful_registration = False
if not pubkey_path.is_file(): if not public_key:
sunnylink_dongle_id = UNREGISTERED_SUNNYLINK_DONGLE_ID sunnylink_dongle_id = UNREGISTERED_SUNNYLINK_DONGLE_ID
self._status_update("Public key not found, setting dongle ID to unregistered.") self._status_update("Public key not found, setting dongle ID to unregistered.")
else: else:
Params().put("LastSunnylinkPingTime", 0) # Reset the last ping time to 0 if we are trying to register Params().put("LastSunnylinkPingTime", 0) # Reset the last ping time to 0 if we are trying to register
with pubkey_path.open() as f1, privkey_path.open() as f2:
public_key = f1.read()
private_key = f2.read()
backoff = 1 backoff = 1
while True: while True:
register_token = jwt.encode({'register': True, 'exp': datetime.utcnow() + timedelta(hours=1)}, private_key, algorithm='RS256') register_token = jwt.encode({'register': True, 'exp': datetime.utcnow() + timedelta(hours=1)}, private_key, algorithm=jwt_algo)
try: try:
if verbose or time.monotonic() - start_time < timeout / 2: if verbose or time.monotonic() - start_time < timeout / 2:
self._status_update("Registering device to sunnylink...") self._status_update("Registering device to sunnylink...")