tomekwojcik/hotpocket
1
0
Fork 0
Code Actions Releases 11 Activity

BTHLABS-50: Safari Web Extension: Reloaded

Browse Source 
Turns out, getting this thing out into the wild isn't as simple as I thought :D
Co-authored-by: Tomek Wójcik <labs@tomekwojcik.pl>
Co-committed-by: Tomek Wójcik <labs@tomekwojcik.pl>
This commit is contained in:
Tomek Wójcik
2025-09-11 15:57:11 +00:00
committed by Tomek Wójcik
parent 67138c7035
commit dcebccf947
15 changed files with 456 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files

39
services/backend/hotpocket_backend/apps/accounts/services/access_tokens.py
View File

@@ -11,7 +11,10 @@ import uuid6
from hotpocket_backend.apps.accounts.models import AccessToken
from hotpocket_backend.apps.core.conf import settings
from hotpocket_soa.dto.accounts import AccessTokensQuery
from hotpocket_soa.dto.accounts import (
AccessTokenMetaUpdateIn,
AccessTokensQuery,
)
LOGGER = logging.getLogger(__name__)
@@ -54,6 +57,16 @@ class AccessTokensService:
f'Access Token not found: pk=`{pk}`',
) from exception
def get_by_key(self, *, key: str) -> AccessToken:
try:
query_set = AccessToken.active_objects
return query_set.get(key=key)
except AccessToken.DoesNotExist as exception:
raise self.AccessTokenNotFound(
f'Access Token not found: key=`{key}`',
) from exception
def search(self,
*,
query: AccessTokensQuery,
@@ -79,3 +92,27 @@ class AccessTokensService:
access_token.soft_delete()
return True
def update_meta(self,
*,
pk: uuid.UUID,
update: AccessTokenMetaUpdateIn,
) -> AccessToken:
access_token = AccessToken.active_objects.get(pk=pk)
next_meta = {
**(access_token.meta or {}),
}
if update.version is not None:
next_meta['version'] = update.version
if update.platform is not None:
next_meta['platform'] = update.platform
access_token.meta = next_meta
access_token.save()
access_token.refresh_from_db()
return access_token

52
services/backend/hotpocket_backend/apps/ui/rpc_methods/accounts/auth.py
View File

@@ -1,10 +1,62 @@
# -*- coding: utf-8 -*-
from __future__ import annotations
import logging
from bthlabs_jsonrpc_core import register_method
from django import db
from django.http import HttpRequest
from hotpocket_soa.dto.accounts import AccessTokenMetaUpdateIn
from hotpocket_soa.services import AccessTokensService
LOGGER = logging.getLogger(__name__)
@register_method('accounts.auth.check')
def check(request: HttpRequest) -> bool:
return request.user.is_anonymous is False
@register_method('accounts.auth.check_access_token')
def check_access_token(request: HttpRequest,
access_token: str,
meta: dict | None = None,
) -> bool:
result = True
try:
access_tokens_service = AccessTokensService()
with db.transaction.atomic():
access_token_object = access_tokens_service.get_by_key(
account_uuid=request.user.pk,
key=access_token,
)
meta_update = AccessTokenMetaUpdateIn.model_validate(
(meta or {}),
)
_ = access_tokens_service.update_meta(
access_token=access_token_object,
update=meta_update,
)
except AccessTokensService.AccessTokenNotFound as exception:
LOGGER.error(
'Access Token not found: account_uuid=`%s` key=`%s`',
request.user.pk,
access_token,
exc_info=exception,
)
result = False
except AccessTokensService.AccessTokenAccessDenied as exception:
LOGGER.error(
'Access Token access denied: account_uuid=`%s` key=`%s`',
request.user.pk,
access_token,
exc_info=exception,
)
result = False
return result

4
services/backend/testing/hotpocket_backend_testing/factories/accounts/access_token.py
View File

@@ -11,7 +11,7 @@ from hotpocket_backend.apps.accounts.models import AccessToken
def AccessTokenMetaFactory() -> dict:
return {
'platform': 'MacIntel',
'version': '1987.10.03',
'version': '1985.12.12',
}
@@ -19,7 +19,7 @@ class AccessTokenFactory(factory.django.DjangoModelFactory):
account_uuid = None
key = factory.LazyFunction(lambda: str(uuid.uuid4()))
origin = factory.LazyFunction(
lambda: f'safari-web-extension//{uuid.uuid4()}',
lambda: f'safari-web-extension://{uuid.uuid4()}',
)
meta = factory.LazyFunction(AccessTokenMetaFactory)

1
services/backend/testing/hotpocket_backend_testing/fixtures/accounts/__init__.py
View File

@@ -1,2 +1,3 @@
from .access_token import * # noqa: F401,F403
from .account import * # noqa: F401,F403
from .apps import * # noqa: F401,F403

20
services/backend/testing/hotpocket_backend_testing/fixtures/accounts/apps.py Normal file
View File

@@ -0,0 +1,20 @@
# -*- coding: utf-8 -*-
# type: ignore
from __future__ import annotations
import uuid
import pytest
@pytest.fixture
def safari_extension_origin():
return f'safari-web-extension://{uuid.uuid4()}'
@pytest.fixture
def safari_extension_meta():
return {
'platform': 'MacIntel',
'version': '1987.10.03',
}

25
services/backend/testing/hotpocket_backend_testing/services/accounts/access_tokens.py
View File

@@ -24,15 +24,28 @@ class AccessTokensTestingService:
assert access_token.updated_at is not None
def assert_deleted(self, *, pk: uuid.UUID, reference: typing.Any = None):
association = AccessToken.objects.get(pk=pk)
assert association.deleted_at is not None
access_token = AccessToken.objects.get(pk=pk)
assert access_token.deleted_at is not None
if reference is not None:
assert association.updated_at > reference.updated_at
assert access_token.updated_at > reference.updated_at
def assert_not_deleted(self, *, pk: uuid.UUID, reference: typing.Any = None):
association = AccessToken.objects.get(pk=pk)
assert association.deleted_at is None
access_token = AccessToken.objects.get(pk=pk)
assert access_token.deleted_at is None
if reference is not None:
assert association.updated_at == reference.updated_at
assert access_token.updated_at == reference.updated_at
def assert_meta_updated(self, *, pk: uuid.UUID, meta_update: dict, reference: typing.Any = None):
access_token = AccessToken.objects.get(pk=pk)
if len(meta_update) > 0 and reference is not None:
expected_meta = {
**reference.meta,
**meta_update,
}
assert access_token.meta == expected_meta
if reference is not None:
assert access_token.updated_at > reference.updated_at

35
services/backend/tests/ui/views/rpc/accounts/access_tokens/test_create.py
View File

@@ -14,29 +14,16 @@ from hotpocket_backend_testing.services.accounts import (
)
@pytest.fixture
def origin():
return f'safari-web-extension://{uuid.uuid4()}'
@pytest.fixture
def auth_key():
return str(uuid.uuid4())
@pytest.fixture
def meta():
return {
'platform': 'MacIntel',
'version': '1987.10.03',
}
@pytest.fixture
def call(rpc_call_factory, auth_key, meta):
def call(rpc_call_factory, auth_key, safari_extension_meta):
return rpc_call_factory(
'accounts.access_tokens.create',
[auth_key, meta],
[auth_key, safari_extension_meta],
)
@@ -44,9 +31,9 @@ def call(rpc_call_factory, auth_key, meta):
def test_ok(authenticated_client: Client,
auth_key,
call,
origin,
safari_extension_origin,
account,
meta,
safari_extension_meta,
):
# Given
session = authenticated_client.session
@@ -59,7 +46,7 @@ def test_ok(authenticated_client: Client,
data=call,
content_type='application/json',
headers={
'Origin': origin,
'Origin': safari_extension_origin,
},
)
@@ -72,8 +59,8 @@ def test_ok(authenticated_client: Client,
AccessTokensTestingService().assert_created(
key=call_result['result'],
account_uuid=account.pk,
origin=origin,
meta=meta,
origin=safari_extension_origin,
meta=safari_extension_meta,
)
assert 'extension_auth_key' not in authenticated_client.session
@@ -82,7 +69,7 @@ def test_ok(authenticated_client: Client,
@pytest.mark.django_db
def test_auth_key_missing(authenticated_client: Client,
call,
origin,
safari_extension_origin,
):
# When
result = authenticated_client.post(
@@ -90,7 +77,7 @@ def test_auth_key_missing(authenticated_client: Client,
data=call,
content_type='application/json',
headers={
'Origin': origin,
'Origin': safari_extension_origin,
},
)
@@ -105,7 +92,7 @@ def test_auth_key_missing(authenticated_client: Client,
@pytest.mark.django_db
def test_auth_key_mismatch(authenticated_client: Client,
call,
origin,
safari_extension_origin,
):
# Given
session = authenticated_client.session
@@ -118,7 +105,7 @@ def test_auth_key_mismatch(authenticated_client: Client,
data=call,
content_type='application/json',
headers={
'Origin': origin,
'Origin': safari_extension_origin,
},
)

205
services/backend/tests/ui/views/rpc/accounts/auth/test_check_access_token.py Normal file
View File

@@ -0,0 +1,205 @@
# -*- coding: utf-8 -*-
# type: ignore
from __future__ import annotations
import http
from django.test import Client
from django.urls import reverse
import pytest
from hotpocket_backend_testing.services.accounts import (
AccessTokensTestingService,
)
@pytest.fixture
def call_factory(request: pytest.FixtureRequest, rpc_call_factory):
default_access_token = request.getfixturevalue('access_token_out')
default_meta_update = request.getfixturevalue('safari_extension_meta')
def factory(access_token=None, meta_update=None):
return rpc_call_factory(
'accounts.auth.check_access_token',
[
(
access_token.key
if access_token is not None
else default_access_token.key
),
(
meta_update
if meta_update is not None
else default_meta_update
),
],
)
return factory
@pytest.fixture
def call(call_factory):
return call_factory()
@pytest.mark.django_db
def test_ok(authenticated_client: Client,
call,
access_token_out,
safari_extension_meta,
):
# When
result = authenticated_client.post(
reverse('ui.rpc'),
data=call,
content_type='application/json',
)
# Then
assert result.status_code == http.HTTPStatus.OK
call_result = result.json()
assert 'error' not in call_result
assert call_result['result'] is True
AccessTokensTestingService().assert_meta_updated(
pk=access_token_out.pk,
meta_update=safari_extension_meta,
reference=access_token_out,
)
@pytest.mark.parametrize(
'meta_keys_to_pop',
[
('platform',),
('version',),
('platform', 'version'),
],
)
@pytest.mark.django_db
def test_ok_with_partial_meta_update(meta_keys_to_pop,
safari_extension_meta,
authenticated_client: Client,
call_factory,
access_token_out,
):
# Given
meta_update = {**safari_extension_meta}
for meta_key_to_pop in meta_keys_to_pop:
meta_update.pop(meta_key_to_pop)
call = call_factory(meta_update=meta_update)
# When
result = authenticated_client.post(
reverse('ui.rpc'),
data=call,
content_type='application/json',
)
# Then
assert result.status_code == http.HTTPStatus.OK
call_result = result.json()
assert 'error' not in call_result
assert call_result['result'] is True
AccessTokensTestingService().assert_meta_updated(
pk=access_token_out.pk,
meta_update=meta_update,
reference=access_token_out,
)
@pytest.mark.django_db
def test_invalid_access_token(authenticated_client: Client,
call,
):
# Given
call['params'][0] = 'thisisntright'
# When
result = authenticated_client.post(
reverse('ui.rpc'),
data=call,
content_type='application/json',
)
# Then
assert result.status_code == http.HTTPStatus.OK
call_result = result.json()
assert 'error' not in call_result
assert call_result['result'] is False
@pytest.mark.django_db
def test_deleted_access_token(call_factory,
deleted_access_token_out,
authenticated_client: Client,
):
# Given
call = call_factory(access_token=deleted_access_token_out)
# When
result = authenticated_client.post(
reverse('ui.rpc'),
data=call,
content_type='application/json',
)
# Then
assert result.status_code == http.HTTPStatus.OK
call_result = result.json()
assert 'error' not in call_result
assert call_result['result'] is False
@pytest.mark.django_db
def test_other_account_access_token(call_factory,
other_account_access_token_out,
authenticated_client: Client,
):
# Given
call = call_factory(access_token=other_account_access_token_out)
# When
result = authenticated_client.post(
reverse('ui.rpc'),
data=call,
content_type='application/json',
)
# Then
assert result.status_code == http.HTTPStatus.OK
call_result = result.json()
assert 'error' not in call_result
assert call_result['result'] is False
@pytest.mark.django_db
def test_inactive_account(inactive_account_client: Client, call):
# When
result = inactive_account_client.post(
reverse('ui.rpc'),
data=call,
)
# Then
assert result.status_code == http.HTTPStatus.FORBIDDEN
@pytest.mark.django_db
def test_anonymous(client: Client, call):
# When
result = client.post(
reverse('ui.rpc'),
data=call,
)
# Then
assert result.status_code == http.HTTPStatus.FORBIDDEN