mirror of
https://github.com/dragonpilot/dragonpilot.git
synced 2026-07-08 06:22:03 +08:00
registration: remove key generation (#23111)
* registration: remove key generation * add test
This commit is contained in:
@@ -1,14 +1,13 @@
|
||||
#!/usr/bin/env python3
|
||||
import os
|
||||
#/!/usr/bin/env python3
|
||||
import time
|
||||
import json
|
||||
import jwt
|
||||
from pathlib import Path
|
||||
|
||||
from datetime import datetime, timedelta
|
||||
from common.api import api_get
|
||||
from common.params import Params
|
||||
from common.spinner import Spinner
|
||||
from common.file_helpers import mkdirs_exists_ok
|
||||
from common.basedir import PERSIST
|
||||
from selfdrive.controls.lib.alertmanager import set_offroad_alert
|
||||
from selfdrive.hardware import HARDWARE
|
||||
@@ -27,19 +26,11 @@ def register(show_spinner=False) -> str:
|
||||
dongle_id = params.get("DongleId", encoding='utf8')
|
||||
needs_registration = None in (IMEI, HardwareSerial, dongle_id)
|
||||
|
||||
# create a key for auth
|
||||
# your private key is kept on your device persist partition and never sent to our servers
|
||||
# do not erase your persist partition
|
||||
if not os.path.isfile(PERSIST+"/comma/id_rsa.pub"):
|
||||
needs_registration = True
|
||||
cloudlog.warning("generating your personal RSA key")
|
||||
mkdirs_exists_ok(PERSIST+"/comma")
|
||||
assert os.system("openssl genrsa -out "+PERSIST+"/comma/id_rsa.tmp 2048") == 0
|
||||
assert os.system("openssl rsa -in "+PERSIST+"/comma/id_rsa.tmp -pubout -out "+PERSIST+"/comma/id_rsa.tmp.pub") == 0
|
||||
os.rename(PERSIST+"/comma/id_rsa.tmp", PERSIST+"/comma/id_rsa")
|
||||
os.rename(PERSIST+"/comma/id_rsa.tmp.pub", PERSIST+"/comma/id_rsa.pub")
|
||||
|
||||
if needs_registration:
|
||||
pubkey = Path(PERSIST+"/comma/id_rsa.pub")
|
||||
if not pubkey.is_file():
|
||||
dongle_id = UNREGISTERED_DONGLE_ID
|
||||
cloudlog.warning(f"missing public key: {pubkey}")
|
||||
elif needs_registration:
|
||||
if show_spinner:
|
||||
spinner = Spinner()
|
||||
spinner.update("registering device")
|
||||
|
||||
@@ -50,6 +50,14 @@ class TestRegistration(unittest.TestCase):
|
||||
self.assertEqual(register(), dongle)
|
||||
self.assertFalse(m.called)
|
||||
|
||||
def test_no_keys(self):
|
||||
# missing pubkey
|
||||
with mock.patch("selfdrive.athena.registration.api_get", autospec=True) as m:
|
||||
dongle = register()
|
||||
self.assertEqual(m.call_count, 0)
|
||||
self.assertEqual(dongle, UNREGISTERED_DONGLE_ID)
|
||||
self.assertEqual(self.params.get("DongleId", encoding='utf-8'), dongle)
|
||||
|
||||
def test_missing_cache(self):
|
||||
# keys exist but no dongle id
|
||||
self._generate_keys()
|
||||
@@ -65,7 +73,8 @@ class TestRegistration(unittest.TestCase):
|
||||
self.assertEqual(self.params.get("DongleId", encoding='utf-8'), dongle)
|
||||
|
||||
def test_unregistered(self):
|
||||
# no keys, no dongle id
|
||||
# keys exist, but unregistered
|
||||
self._generate_keys()
|
||||
with mock.patch("selfdrive.athena.registration.api_get", autospec=True) as m:
|
||||
m.return_value = MockResponse(None, 402)
|
||||
dongle = register()
|
||||
|
||||
Reference in New Issue
Block a user