Prevent non-admin issue responses exposing account identities
Magent CI/CD / verify (push) Successful in 10m34s
Magent CI/CD / deploy-prod (push) Skipped
Magent CI/CD / deploy-beta (push) Skipped

This commit is contained in:
2026-09-07 19:45:19 +12:00
parent 131b5fc5c7
commit 13edcb8136
3 changed files with 74 additions and 3 deletions
+40 -2
View File
@@ -1,6 +1,7 @@
from __future__ import annotations
import logging
import re
import time
from datetime import datetime, timezone
from typing import Any, Dict, Optional, Tuple
@@ -18,7 +19,8 @@ from ..db import (
delete_portal_item,
get_portal_item,
get_portal_overview,
list_portal_comments,
list_portal_comments as _list_portal_comments,
get_all_users,
list_portal_item_activity,
list_portal_items,
update_portal_item,
@@ -481,6 +483,31 @@ def _public_media_status_payload(
}
def _public_text(value: Any) -> Any:
if not isinstance(value, str):
return value
identities = {str(u.get(key) or '').strip() for u in get_all_users() for key in ('username', 'email')}
identities.discard('')
if identities:
pattern = r'(?<![\w@])(?:' + '|'.join(re.escape(v) for v in sorted(identities, key=len, reverse=True)) + r')(?![\w@])'
value = re.sub(pattern, '[private]', value, flags=re.IGNORECASE)
return re.sub(r'[\w.+%-]+@[\w.-]+\.[A-Za-z]{2,}', '[private email]', value)
def _public_comment(comment: Dict[str, Any]) -> Dict[str, Any]:
return {
'id': comment.get('id'), 'item_id': comment.get('item_id'),
'author_username': 'Support team' if comment.get('author_role') == 'admin' else 'Reporter',
'author_role': comment.get('author_role'), 'created_at': comment.get('created_at'),
'message': _public_text(comment.get('message')), 'is_internal': False,
}
def list_portal_comments(*args, **kwargs):
comments = _list_portal_comments(*args, **kwargs)
return comments if kwargs.get('include_internal') else [_public_comment(c) for c in comments if not c.get('is_internal')]
def _serialize_item(item: Dict[str, Any], user: Dict[str, Any]) -> Dict[str, Any]:
is_admin = _is_admin(user)
is_owner = _is_owner(user, item)
@@ -525,6 +552,16 @@ def _serialize_item(item: Dict[str, Any], user: Dict[str, Any]) -> Dict[str, Any
"last_delivery_succeeded": resolution.get("lastDeliverySucceeded"),
},
}
if not is_admin:
serialized = {key: value for key, value in serialized.items() if key in {
'id', 'kind', 'title', 'description', 'media_type', 'year', 'source_request_id',
'related_item_id', 'status', 'workflow_request_status', 'workflow_media_status',
'issue_type', 'issue_resolved_at', 'priority', 'created_at', 'updated_at',
'last_activity_at', 'permissions', 'workflow', 'issue',
}}
serialized['created_by_username'] = user.get('username') if is_owner else 'Another member'
serialized['title'] = _public_text(serialized.get('title'))
serialized['description'] = _public_text(serialized.get('description'))
return serialized
@@ -566,6 +603,7 @@ def _activity_payload(item: Dict[str, Any], *, include_internal: bool = False) -
else "Reporter"
)
public_entry["actor_role"] = "system" if actor_role == "system" else "support" if actor_role == "admin" else "user"
public_entry['message'] = _public_text(public_entry.get('message'))
public_activity.append(public_entry)
return public_activity
@@ -1510,4 +1548,4 @@ async def portal_create_comment(
user=current_user,
note=f"internal={is_internal}",
)
return {"comment": comment}
return {"comment": comment if is_admin else _public_comment(comment)}
+33
View File
@@ -0,0 +1,33 @@
import unittest
from unittest.mock import patch
from fastapi import FastAPI
from fastapi.testclient import TestClient
from backend.app.routers import portal
class PortalPrivacyTests(unittest.TestCase):
def test_detail_and_comments_are_private_for_regular_users(self):
item = {'id': 1, 'kind': 'issue', 'title': 'Broken movie', 'status': 'new',
'created_by_username': 'private-reporter', 'created_by_id': 42,
'assignee_username': 'private-admin', 'metadata_json': '{"email":"secret@example.com"}',
'description': 'Contact private-reporter or secret@example.com', 'created_at': '2026-09-07'}
comment = {'id': 1, 'item_id': 1, 'author_username': 'private-admin', 'author_role': 'admin',
'message': 'Sent to secret@example.com for private-reporter', 'is_internal': False}
app = FastAPI()
app.include_router(portal.router)
app.dependency_overrides[portal.get_current_user] = lambda: {'username': 'viewer', 'role': 'user'}
with patch.object(portal, 'get_portal_item', return_value=item), \
patch.object(portal, '_list_portal_comments', return_value=[comment]), \
patch.object(portal, 'list_portal_item_activity', return_value=[]), \
patch.object(portal, 'issue_resolution_state', return_value={}), \
patch.object(portal, 'get_all_users', return_value=[{'username': 'private-reporter', 'email': 'secret@example.com'}, {'username': 'private-admin'}]):
client = TestClient(app)
for path in ['/portal/items/1', '/portal/items/1/comments']:
response = client.get(path)
self.assertEqual(response.status_code, 200)
for secret in ['private-reporter', 'private-admin', 'secret@example.com', 'metadata_json', 'assignee_username', 'created_by_id']:
self.assertNotIn(secret, response.text)
admin_result = portal._serialize_item(item, {'username': 'admin', 'role': 'admin'})
self.assertEqual(admin_result['created_by_username'], 'private-reporter')
own_result = portal._serialize_item(item, {'username': 'private-reporter', 'role': 'user'})
self.assertTrue(own_result['permissions']['can_edit'])