j3d1/toolshed
2
1
Fork 0
Code Issues 9 Pull requests 1 Projects Releases Packages Wiki Activity

stash

Browse source
This commit is contained in:
j3d1 2026-08-01 16:03:35 +02:00
parent 2f8683add1
commit cfcc2c15d3
15 changed files with 644 additions and 77 deletions
Download patch file Download diff file Expand all files Collapse all files

31
backend/authentication/api.py
View file

@ -10,7 +10,8 @@ from rest_framework.authtoken.views import ObtainAuthToken
from rest_framework.response import Response
from authentication.models import ToolshedUser, AccountPreference
from authentication.signature_auth import SignatureAuthenticationLocal
from authentication.signature_auth import SignatureAuthenticationLocal, SignatureAuthentication, \
split_userhandle_or_throw
from files.models import File
from files.serializers import FileSerializer
from hostadmin.models import Domain
@ -101,6 +102,9 @@ class UserViewSet(viewsets.ModelViewSet):
@permission_classes([IsAuthenticated])
@authentication_classes([SignatureAuthenticationLocal])
def getUserInfo(request):
"""Get or update the authenticated local user's own account info. Only usable by the
account owner on their own home server - see getUserProfile for viewing another (friend)
user's public profile."""
user = request.user
if request.method == 'PATCH':
old_file = user.profile_picture
@ -137,6 +141,30 @@ def getUserInfo(request):
})
@api_view(['GET'])
@permission_classes([IsAuthenticated])
@authentication_classes([SignatureAuthentication])
def getUserProfile(request, handle):
"""Get another local user's public profile by handle (username@domain), e.g. so a friend
can look up someone's avatar. The caller must be a friend of that user (or the user
itself, signing with their own known identity rather than their local credentials)."""
try:
username, domain = split_userhandle_or_throw(handle)
except ValueError:
return Response(status=400)
try:
target = ToolshedUser.objects.get(username=username, domain=domain)
except ToolshedUser.DoesNotExist:
return Response(status=404)
if target not in request.user.friends_or_self():
return Response(status=403)
return Response({
'username': target.username,
'domain': target.domain,
'profile_picture': FileSerializer(target.profile_picture).data if target.profile_picture else None,
})
@api_view(['POST'])
@permission_classes([])
@authentication_classes([])
@ -212,6 +240,7 @@ router.register(r'users', UserViewSet)
urlpatterns = [
path('', include(router.urls)),
path('user/', getUserInfo),
path('user/<str:handle>/', getUserProfile),
path('register/', registerUser),
path('token/', UserAuthToken.as_view()),
path('preferences/', preference_definitions),

16
backend/authentication/signature_auth.py
View file

@ -106,11 +106,19 @@ def authenticate_request_against_local_users(request, raw_request_body):
class SignatureAuthentication(authentication.BaseAuthentication):
def authenticate(self, request):
return authenticate_request_against_known_identities(
request, request.body.decode('utf-8')), None
identity = authenticate_request_against_known_identities(request, request.body.decode('utf-8'))
# Returning a bare None (rather than a (None, None) tuple) tells DRF this
# authenticator doesn't apply, so it moves on to the next authenticator in the
# authentication_classes list instead of treating the request as authenticated
# with an empty user.
if identity is None:
return None
return identity, None
class SignatureAuthenticationLocal(authentication.BaseAuthentication):
def authenticate(self, request):
return authenticate_request_against_local_users(
request, request.body.decode('utf-8')), None
user = authenticate_request_against_local_users(request, request.body.decode('utf-8'))
if user is None:
return None
return user, None

46
backend/authentication/tests/test_auth.py
View file

@ -355,6 +355,52 @@ class UserApiTestCase(UserTestMixin, ToolshedTestCase):
self.assertEqual(reply.status_code, 403)
class UserProfileByHandleApiTestCase(UserTestMixin, ToolshedTestCase):
"""Tests for GET /auth/user/<handle>/ - viewing another (friend) user's public profile."""
def setUp(self):
super().setUp()
self.prepare_users()
self.f['local_user1'].friends.add(self.f['ext_user1'].public_identity)
self.anonymous_client = Client(SERVER_NAME='testserver')
self.client = SignatureAuthClient()
def test_view_friend_profile(self):
target = '/auth/user/' + str(self.f['local_user1']) + '/'
reply = self.client.get(target, self.f['ext_user1'])
self.assertEqual(reply.status_code, 200)
self.assertEqual(reply.json()['username'], 'testuser1')
self.assertEqual(reply.json()['domain'], 'example.com')
self.assertIsNone(reply.json()['profile_picture'])
self.assertNotIn('email', reply.json())
def test_view_own_profile_via_handle(self):
target = '/auth/user/' + str(self.f['local_user1']) + '/'
reply = self.client.get(target, self.f['local_user1'])
self.assertEqual(reply.status_code, 200)
self.assertEqual(reply.json()['username'], 'testuser1')
def test_view_profile_not_friend(self):
target = '/auth/user/' + str(self.f['local_user1']) + '/'
reply = self.client.get(target, self.f['ext_user2'])
self.assertEqual(reply.status_code, 403)
def test_view_profile_unknown_user(self):
target = '/auth/user/nosuchuser@example.com/'
reply = self.client.get(target, self.f['ext_user1'])
self.assertEqual(reply.status_code, 404)
def test_view_profile_bad_handle(self):
target = '/auth/user/notahandle/'
reply = self.client.get(target, self.f['ext_user1'])
self.assertEqual(reply.status_code, 400)
def test_view_profile_unauthenticated(self):
target = '/auth/user/' + str(self.f['local_user1']) + '/'
reply = self.anonymous_client.get(target)
self.assertEqual(reply.status_code, 403)
class FriendApiTestCase(UserTestMixin, ToolshedTestCase):
def setUp(self):
super().setUp()

10
backend/files/media_urls.py
View file

@ -17,6 +17,12 @@ from files.models import File
@permission_classes([IsAuthenticated])
@authentication_classes([SignatureAuthentication])
def media_urls(request, hash_path):
# Note: CORS headers are NOT set here - django-cors-headers (CorsMiddleware,
# configured in settings.py) adds them to every Django response automatically, so
# setting them manually on these responses would just be redundant. The one exception
# is the SERVE_X_ACCEL_REDIRECT path: nginx replaces this response entirely when it
# follows the X-Accel-Redirect and serves the file itself, so the CORS header for that
# case has to be configured in nginx's `location /redirect_media/` block instead.
try:
file = File.objects.filter(
Q(connected_items__owner__in=request.user.friends_or_self()) |
@ -29,14 +35,10 @@ def media_urls(request, hash_path):
content_type=file.mime_type,
headers={
'X-Accel-Redirect': f'/redirect_media/{hash_path}',
'Access-Control-Allow-Origin': '*',
}) # TODO Expires and Cache-Control
else:
return HttpResponse(status=status.HTTP_200_OK,
content_type=file.mime_type,
headers={
'Access-Control-Allow-Origin': '*',
},
content=open(file.file.path, 'rb').read())

17
backend/toolshed/api/inventory.py
View file

@ -22,7 +22,8 @@ def inventory_items(identity):
except ToolshedUser.DoesNotExist:
pass
for friend in identity.friends.all():
if friend_user := friend.user.get():
friend_user = friend.user.first()
if friend_user:
for item in friend_user.inventory_items.all():
if item.availability_policy != 'private':
yield item
@ -52,13 +53,25 @@ class InventoryItemViewSet(viewsets.ModelViewSet):
instance.delete()
def matches_query(item, query):
query = query.lower()
if query in item.name.lower():
return True
if item.description and query in item.description.lower():
return True
if any(query in tag.name.lower() for tag in item.tags.all()):
return True
return False
@api_view(['GET'])
@authentication_classes([SignatureAuthentication])
@permission_classes([IsAuthenticated])
def search_inventory_items(request):
query = request.query_params.get('query')
if query:
return Response(InventoryItemSerializer(inventory_items(request.user), many=True).data)
matching_items = [item for item in inventory_items(request.user) if matches_query(item, query)]
return Response(InventoryItemSerializer(matching_items, many=True).data)
return Response({'error': 'No query provided.'}, status=400)

6
backend/toolshed/serializers.py
View file

@ -94,7 +94,7 @@ class CategorySerializer(serializers.ModelSerializer):
return obj.get_handle()
def to_representation(self, instance):
return instance.get_handle()
return instance.name
def to_internal_value(self, data):
return resolve_category_handle(data.split("/")[-1])
@ -129,7 +129,7 @@ class ItemPropertySerializer(serializers.ModelSerializer):
return obj.property.get_handle()
def to_representation(self, instance):
return {'value': instance.value, 'name': instance.property.name, 'handle': instance.property.get_handle()}
return {'value': instance.value, 'name': instance.property.name}
def to_internal_value(self, data):
prop = resolve_property_handle(data.get('name') or data.get('handle'))
@ -151,7 +151,7 @@ class InventoryItemSerializer(serializers.ModelSerializer):
'tags', 'tags_input', 'properties', 'files', 'storage_location']
def get_tags(self, obj):
return [tag.get_handle() for tag in obj.tags.all()]
return [tag.name for tag in obj.tags.all()]
def to_internal_value(self, data):
files = data.pop('files', [])

264
backend/toolshed/tests/test_offlinedata.py Normal file
View file

@ -0,0 +1,264 @@
from django.core.files.base import ContentFile
from django.test import Client
from authentication.models import AccountPreference, ToolshedUser
from authentication.tests import UserTestMixin, SignatureAuthClient, ToolshedTestCase
from files.models import File
from toolshed.models import InventoryItem, ItemProperty, StorageLocation
from toolshed.offlinedata import import_inventory, inventory_rows, rows_to_csv
from toolshed.tests import CategoryTestMixin, LocationTestMixin, PropertyTestMixin, TagTestMixin
anonymous_client = Client()
client = SignatureAuthClient()
class _DeleteTestDataMixin(UserTestMixin, CategoryTestMixin, LocationTestMixin):
"""Shared fixture setup for the delete-data and delete-account test cases."""
def setUp(self):
super().setUp()
self.prepare_users()
self.prepare_categories()
self.prepare_locations()
self.f['local_user1'].friends.add(self.f['local_user2'].public_identity)
self.f['shared_file'] = File.objects.create(
file=ContentFile(b'shared', 'shared'), mime_type='text/plain', hash='shared')
self.f['orphan_file'] = File.objects.create(
file=ContentFile(b'orphan', 'orphan'), mime_type='text/plain', hash='orphan')
self.f['item1'] = InventoryItem.objects.create(
owner=self.f['local_user1'], owned_quantity=1, name='item1', category=self.f['cat1'])
self.f['item1'].files.add(self.f['orphan_file'])
self.f['item_other_user'] = InventoryItem.objects.create(
owner=self.f['local_user2'], owned_quantity=1, name='item2', category=self.f['cat1'])
self.f['item_other_user'].files.add(self.f['shared_file'])
self.f['item1'].files.add(self.f['shared_file'])
AccountPreference.objects.create(user=self.f['local_user1'], key='theme', value='dark')
self.f['local_user1'].profile_picture = self.f['orphan_file']
self.f['local_user1'].save()
class DeleteDataTestCase(_DeleteTestDataMixin, ToolshedTestCase):
def test_delete_data_anonymous(self):
response = anonymous_client.delete('/api/account_data/')
self.assertEqual(response.status_code, 403)
def test_delete_data_removes_all_owned_data_but_keeps_account(self):
response = client.delete('/api/account_data/', self.f['local_user1'])
self.assertEqual(response.status_code, 200)
summary = response.json()
self.assertEqual(summary['inventory_items'], 1)
self.assertEqual(summary['locations'], 4)
self.assertEqual(summary['settings'], 1)
self.assertEqual(summary['friends'], 1)
# the account itself survives - this wipes data, it doesn't close the account
self.f['local_user1'].refresh_from_db()
self.assertTrue(ToolshedUser.objects.filter(username='testuser1').exists())
self.assertIsNone(self.f['local_user1'].profile_picture)
self.assertFalse(InventoryItem.global_objects.filter(owner_id=self.f['local_user1'].id).exists())
self.assertFalse(StorageLocation.objects.filter(owner_id=self.f['local_user1'].id).exists())
self.assertFalse(AccountPreference.objects.filter(user_id=self.f['local_user1'].id).exists())
self.assertEqual(self.f['local_user1'].public_identity.friends.count(), 0)
# orphaned file (only referenced by the deleted user/items) is gone
self.assertFalse(File.objects.filter(hash='orphan').exists())
# file still referenced by the other user's item survives
self.assertTrue(File.objects.filter(hash='shared').exists())
# the other user's data and identity/friend relation to the deleted identity are untouched
self.f['local_user2'].refresh_from_db()
self.assertTrue(InventoryItem.objects.filter(owner=self.f['local_user2']).exists())
class DeleteAccountTestCase(_DeleteTestDataMixin, ToolshedTestCase):
def test_delete_account_anonymous(self):
response = anonymous_client.delete('/api/account/')
self.assertEqual(response.status_code, 403)
def test_delete_account_removes_data_and_closes_account(self):
user1_id = self.f['local_user1'].id
identity_id = self.f['local_user1'].public_identity_id
response = client.delete('/api/account/', self.f['local_user1'])
self.assertEqual(response.status_code, 200)
summary = response.json()
self.assertEqual(summary['inventory_items'], 1)
self.assertEqual(summary['locations'], 4)
self.assertEqual(summary['settings'], 1)
self.assertEqual(summary['friends'], 1)
self.assertTrue(summary['account'])
# the account itself is gone
self.assertFalse(ToolshedUser.objects.filter(id=user1_id).exists())
self.assertFalse(InventoryItem.global_objects.filter(owner_id=user1_id).exists())
self.assertFalse(StorageLocation.objects.filter(owner_id=user1_id).exists())
self.assertFalse(AccountPreference.objects.filter(user_id=user1_id).exists())
# the underlying identity is kept, so remote friends/history referencing it stay intact
from authentication.models import KnownIdentity
self.assertTrue(KnownIdentity.objects.filter(id=identity_id).exists())
# orphaned file (only referenced by the deleted user/items) is gone
self.assertFalse(File.objects.filter(hash='orphan').exists())
# file still referenced by the other user's item survives
self.assertTrue(File.objects.filter(hash='shared').exists())
# the other user's data and identity/friend relation to the deleted identity are untouched
self.f['local_user2'].refresh_from_db()
self.assertTrue(InventoryItem.objects.filter(owner=self.f['local_user2']).exists())
self.assertEqual(self.f['local_user2'].public_identity.friends.count(), 0)
class ImportInventoryPropertiesTestCase(UserTestMixin, CategoryTestMixin, TagTestMixin, PropertyTestMixin,
ToolshedTestCase):
"""Properties must round-trip through export/import even when their value contains a
comma or an '=' sign - characters that a naive "handle=value, handle2=value2" encoding of
the 'properties' CSV cell would misinterpret as a field/entry separator.
"""
def setUp(self):
super().setUp()
self.prepare_users()
self.prepare_categories()
self.prepare_tags()
self.prepare_properties()
def test_property_values_with_comma_and_equals_round_trip(self):
item = InventoryItem.objects.create(owner=self.f['local_user1'], name='widget')
ItemProperty.objects.create(inventory_item=item, property=self.f['prop1'], value='10cm, 20cm')
ItemProperty.objects.create(inventory_item=item, property=self.f['prop2'], value='a=b')
csv_bytes = b''.join(rows_to_csv(list(inventory_rows(self.f['local_user1']))))
imported, errors = import_inventory(self.f['local_user2'], csv_bytes, available_files={})
self.assertEqual(errors, [])
self.assertEqual(imported, 1)
new_item = InventoryItem.objects.get(owner=self.f['local_user2'], name='widget')
values = {ip.property.name: ip.value for ip in new_item.itemproperty_set.select_related('property')}
self.assertEqual(values, {'prop1': '10cm, 20cm', 'prop2': 'a=b'})
def test_legacy_comma_equals_format_is_still_importable(self):
handle1 = self.f['prop1'].get_handle()
handle2 = self.f['prop2'].get_handle()
csv_data = (
'name,properties\r\n'
f'legacy widget,"{handle1}=value1, {handle2}=value2"\r\n'
).encode('utf-8')
imported, errors = import_inventory(self.f['local_user1'], csv_data, available_files={})
self.assertEqual(errors, [])
self.assertEqual(imported, 1)
item = InventoryItem.objects.get(owner=self.f['local_user1'], name='legacy widget')
values = {ip.property.name: ip.value for ip in item.itemproperty_set.select_related('property')}
self.assertEqual(values, {'prop1': 'value1', 'prop2': 'value2'})
def test_item_without_properties_imports_cleanly(self):
item = InventoryItem.objects.create(owner=self.f['local_user1'], name='bare item')
csv_bytes = b''.join(rows_to_csv(list(inventory_rows(self.f['local_user1']))))
imported, errors = import_inventory(self.f['local_user2'], csv_bytes, available_files={})
self.assertEqual(errors, [])
self.assertEqual(imported, 1)
new_item = InventoryItem.objects.get(owner=self.f['local_user2'], name='bare item')
self.assertEqual(list(new_item.itemproperty_set.all()), [])
def test_category_and_tags_round_trip(self):
item = InventoryItem.objects.create(
owner=self.f['local_user1'], name='cat and tags item', category=self.f['cat1'])
item.tags.add(self.f['tag1'], self.f['tag2'], through_defaults={})
csv_bytes = b''.join(rows_to_csv(list(inventory_rows(self.f['local_user1']))))
imported, errors = import_inventory(self.f['local_user2'], csv_bytes, available_files={})
self.assertEqual(errors, [])
self.assertEqual(imported, 1)
new_item = InventoryItem.objects.get(owner=self.f['local_user2'], name='cat and tags item')
self.assertEqual(new_item.category, self.f['cat1'])
self.assertEqual(sorted(t.name for t in new_item.tags.all()), ['tag1', 'tag2'])
def test_unknown_property_handle_skips_item_with_error(self):
csv_data = (
'name,properties\r\n'
'ghost widget,test#property:doesnotexist=x\r\n'
).encode('utf-8')
imported, errors = import_inventory(self.f['local_user1'], csv_data, available_files={})
self.assertEqual(imported, 0)
self.assertEqual(len(errors), 1)
self.assertIn('doesnotexist', errors[0])
self.assertFalse(InventoryItem.objects.filter(owner=self.f['local_user1'], name='ghost widget').exists())
def test_property_value_with_quote_character_round_trips(self):
item = InventoryItem.objects.create(owner=self.f['local_user1'], name='quoted widget')
ItemProperty.objects.create(inventory_item=item, property=self.f['prop1'], value='12" screen')
csv_bytes = b''.join(rows_to_csv(list(inventory_rows(self.f['local_user1']))))
imported, errors = import_inventory(self.f['local_user2'], csv_bytes, available_files={})
self.assertEqual(errors, [])
self.assertEqual(imported, 1)
new_item = InventoryItem.objects.get(owner=self.f['local_user2'], name='quoted widget')
values = {ip.property.name: ip.value for ip in new_item.itemproperty_set.select_related('property')}
self.assertEqual(values, {'prop1': '12" screen'})
class ExportImportApiRoundTripTestCase(UserTestMixin, CategoryTestMixin, TagTestMixin, PropertyTestMixin,
ToolshedTestCase):
"""End-to-end coverage of the /api/export/ + /api/import/ endpoints (as actually used by
clients), rather than calling the internal helper functions directly - this is what a real
export/import round trip between two accounts looks like.
"""
def setUp(self):
super().setUp()
self.prepare_users()
self.prepare_categories()
self.prepare_tags()
self.prepare_properties()
def test_export_then_import_preserves_category_tags_and_properties(self):
import base64
item = InventoryItem.objects.create(
owner=self.f['local_user1'], name='drill', description='cordless drill',
category=self.f['cat1'], availability_policy='friends', owned_quantity=2)
item.tags.add(self.f['tag1'], self.f['tag2'], through_defaults={})
ItemProperty.objects.create(inventory_item=item, property=self.f['prop1'], value='10cm, 20cm')
ItemProperty.objects.create(inventory_item=item, property=self.f['prop2'], value='a=b')
export_reply = client.get('/api/export/', self.f['local_user1'])
self.assertEqual(export_reply.status_code, 200)
zip_bytes = export_reply.content
import_reply = client.post('/api/import/', self.f['local_user2'],
{'zip': base64.b64encode(zip_bytes).decode('ascii')})
self.assertEqual(import_reply.status_code, 200)
summary = import_reply.json()
self.assertEqual(summary['inventory_items'], 1)
self.assertEqual(summary['errors'], [])
new_item = InventoryItem.objects.get(owner=self.f['local_user2'], name='drill')
self.assertEqual(new_item.category, self.f['cat1'])
self.assertEqual(sorted(t.name for t in new_item.tags.all()), ['tag1', 'tag2'])
values = {ip.property.name: ip.value for ip in new_item.itemproperty_set.select_related('property')}
self.assertEqual(values, {'prop1': '10cm, 20cm', 'prop2': 'a=b'})