feat: multi-tenant MQTT and strict device isolation#128
Merged
Conversation
Contributor
Reviewer's GuideImplements multi-tenant device and MQTT handling by scoping Firebase paths, MQTT topics, and logs to per-user identifiers, adds a dashboard-friendly device view, and wires routes and tests to enforce strict user isolation and dynamic monitoring. Sequence diagram for multi-tenant smarthome intent handlingsequenceDiagram
actor User
participant GoogleAssistant as GoogleAssistant
participant FlaskApp as FlaskApp
participant Routes as routes_smarthome
participant Actions as action_devices_actions
participant Firebase as Firebase_RTD
participant MQTT as MQTT_Broker
User ->> GoogleAssistant: Send SYNC/QUERY/EXECUTE intent
GoogleAssistant ->> FlaskApp: HTTP POST /smarthome (JSON)
FlaskApp ->> Routes: smarthome()
Routes ->> Routes: _resolve_smarthome_user_scope(req)
note over Routes: Resolve user_scope and agent_user_id
Routes ->> Actions: actions(req, user_id=user_scope, agent_user_id=agent_user_id)
alt SYNC intent
Actions ->> Actions: onSync(user_id, agent_user_id)
Actions ->> Firebase: rsync(user_id)
Firebase -->> Actions: user_scoped_devices
Actions -->> Routes: SYNC payload (agentUserId, devices)
else QUERY intent
Actions ->> Actions: onQuery(body, user_id)
loop for each device
Actions ->> Actions: rquery(deviceId, user_id)
Actions ->> Firebase: _get_scoped_device_states(deviceId, user_id)
Firebase -->> Actions: states
end
Actions -->> Routes: QUERY payload
else EXECUTE intent
Actions ->> Actions: onExecute(body, user_id)
loop for each execution
Actions ->> Actions: commands(payload, deviceId, execCommand, params, user_id)
Actions ->> Actions: rexecute(deviceId, params, user_id)
Actions ->> Firebase: update user_scoped states
Firebase -->> Actions: updated states
end
Actions -->> Routes: EXECUTE payload
note over Actions,MQTT: Publish multi-tenant MQTT notification
Actions ->> MQTT: publish(topic="user_id/deviceId/notification", payload=params)
end
Routes ->> FlaskApp: Build response {'requestId', 'payload'}
FlaskApp -->> GoogleAssistant: JSON response
GoogleAssistant -->> User: Present result
Sequence diagram for multi-tenant MQTT status handling and loggingsequenceDiagram
participant Device as IoT_Device
participant MQTT as MQTT_Broker
participant FlaskMQTT as Flask_MQTT_Client
participant Notifications as notifications_module
participant Firebase as Firebase_RTD
participant Routes as routes_mqtt_log
participant Browser as Web_Browser
Device ->> MQTT: PUBLISH {user_id}/{device_id}/status
MQTT ->> FlaskMQTT: Deliver message
FlaskMQTT ->> Notifications: handle_messages(message)
Notifications ->> Notifications: _decode_payload(payload)
Notifications ->> Notifications: Parse topic into user_id, device_id, msg_type
alt msg_type is status
Notifications ->> Firebase: db.reference(/users/user_id/devices/device_id/states).update(state_updates)
Firebase -->> Notifications: ack
end
Notifications ->> Notifications: _append_mqtt_log(topic, payload, Received, user_id)
User ->> Browser: Open /mqtt_log
Browser ->> Routes: HTTP GET /mqtt_log
Routes ->> Notifications: get_mqtt_logs(user_id)
Notifications ->> Notifications: filter logs by user_id or None
Notifications -->> Routes: filtered_log_entries
Routes -->> Browser: Render mqtt_log.html with logs
Browser ->> User: Show user_scoped MQTT log
Class diagram for updated multi-tenant helpers and MQTT loggingclassDiagram
class ActionDevices {
+reference(user_id)
+rstate(user_id)
+rsync(user_id)
+get_dashboard_devices(user_id)
+rquery(deviceId, user_id)
+rexecute(deviceId, parameters, user_id)
+onSync(user_id, agent_user_id)
+onQuery(body, user_id)
+onExecute(body, user_id)
+commands(payload, deviceId, execCommand, params, user_id)
+actions(req, user_id, agent_user_id)
+report_state(user_id)
_normalize_user_scope(user_id)
_to_sync_device(device_id, raw_data)
_build_sync_devices(snapshot)
_get_scoped_snapshot(user_id)
_get_scoped_device_states(device_id, user_id)
_normalize_device_type(device_type)
}
class RoutesModule {
+sync_devices()
+devices()
+device_status(device_id)
+mqtt_log()
+smarthome()
_resolve_smarthome_user_scope(req)
}
class NotificationsModule {
+get_mqtt_logs(user_id)
+handle_connect(client, userdata, flags, rc)
+handle_disconnect(client, userdata, rc)
+handle_messages(client, userdata, message)
+handle_publish(client, userdata, mid)
_append_mqtt_log(topic, payload, status, user_id)
_decode_payload(payload)
}
class MQTTBroker {
+publish(topic, payload, qos)
+subscribe(topic)
}
class FirebaseRTDB {
+reference(path)
+get()
+child(key)
+update(data)
}
RoutesModule --> ActionDevices : uses
RoutesModule --> NotificationsModule : uses
ActionDevices --> FirebaseRTDB : reads_writes
NotificationsModule --> FirebaseRTDB : updates_status_states
NotificationsModule --> MQTTBroker : subscribes
ActionDevices --> MQTTBroker : publishes_notifications
class DeviceDashboardViewModel {
+id
+display_name
+type
+type_label
+icon
+is_online
+status
+status_class
}
ActionDevices --> DeviceDashboardViewModel : builds_list_for_devices_template
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Code Review
This pull request implements multi-tenancy support by scoping Firebase device data and MQTT communication by user ID. Key changes include updating Firebase references to use user-specific paths, modifying MQTT topics to a multi-tenant structure, and enhancing the smart home intent handler to resolve user scope from authentication tokens or request payloads. Feedback identifies performance improvements for the MQTT callback, the need for better validation and backward compatibility in MQTT message handling, and a potential issue with null user IDs during topic construction.
Comment on lines
+109
to
+112
| from firebase_admin import db | ||
| # Update status in /users/{user_id}/devices/{device_id}/states | ||
| # Assuming payload is a JSON string of states | ||
| import json |
Contributor
There was a problem hiding this comment.
Avoid importing modules inside a function that is called frequently, such as an MQTT message callback. These imports should be moved to the top of the file to improve performance and follow standard Python practices.
Comment on lines
+101
to
+117
| if len(parts) >= 3: | ||
| user_id = parts[0] | ||
| device_id = parts[1] | ||
| msg_type = parts[2] | ||
|
|
||
| # Try to update Firebase if it's a status message | ||
| if msg_type == 'status': | ||
| try: | ||
| from firebase_admin import db | ||
| # Update status in /users/{user_id}/devices/{device_id}/states | ||
| # Assuming payload is a JSON string of states | ||
| import json | ||
| state_updates = json.loads(payload) | ||
| db.reference(f'/users/{user_id}/devices/{device_id}/states').update(state_updates) | ||
| logger.debug("Updated Firebase status for %s/%s", user_id, device_id) | ||
| except Exception as e: | ||
| logger.error("Failed to update Firebase from MQTT: %s", e) |
Contributor
There was a problem hiding this comment.
The logic for updating Firebase from MQTT status messages has several issues:
- Compatibility: It only handles 3-part topics, breaking backward compatibility for legacy 2-part topics (
device/status) which are still subscribed to inapp.py. - Robustness: It should verify that
state_updatesis a dictionary before calling.update()to prevent runtime errors if the payload is a simple JSON value (like a boolean or string). - Sanitization: The
user_idanddevice_idextracted from the topic should be validated/normalized before being used in a Firebase path to prevent potential path traversal if the MQTT broker is not strictly locked down.
| params = payload['commands'][0]['states'] | ||
| mqtt.publish(topic=str(deviceId) + '/' + 'notification', | ||
| # Multi-tenant MQTT topic: {user_id}/{device_id}/notification | ||
| mqtt_topic = f"{user_id}/{deviceId}/notification" |
Contributor
There was a problem hiding this comment.
If user_id is None (e.g., for legacy users or when fallback is active), the MQTT topic will be constructed as None/{deviceId}/notification, which is likely incorrect. It should fall back to the legacy topic structure or use a default prefix to maintain compatibility.
Suggested change
| mqtt_topic = f"{user_id}/{deviceId}/notification" | |
| mqtt_topic = f"{user_id}/{deviceId}/notification" if user_id else f"{deviceId}/notification" |
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Hey - I've found 3 issues, and left some high level feedback:
- In
actionsthe MQTT topic is built directly fromuser_id(f"{user_id}/{deviceId}/notification") without reusing_normalize_user_scope, which means a malicious or malformedagentUserIdfrom the request could inject slashes into the topic hierarchy; consider normalizing or validatinguser_idbefore using it in MQTT topic names. - In
notifications.handle_messages, the Firebase update path/users/{user_id}/devices/{device_id}/statesis constructed inline instead of going through the new helper functions inaction_devices(e.g.,_normalize_user_scope/_get_scoped_device_states), which can lead to inconsistent path validation rules between MQTT-driven updates and HTTP-driven operations; consider centralizing this path construction/normalization.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- In `actions` the MQTT topic is built directly from `user_id` (`f"{user_id}/{deviceId}/notification"`) without reusing `_normalize_user_scope`, which means a malicious or malformed `agentUserId` from the request could inject slashes into the topic hierarchy; consider normalizing or validating `user_id` before using it in MQTT topic names.
- In `notifications.handle_messages`, the Firebase update path `/users/{user_id}/devices/{device_id}/states` is constructed inline instead of going through the new helper functions in `action_devices` (e.g., `_normalize_user_scope` / `_get_scoped_device_states`), which can lead to inconsistent path validation rules between MQTT-driven updates and HTTP-driven operations; consider centralizing this path construction/normalization.
## Individual Comments
### Comment 1
<location path="action_devices.py" line_range="379-380" />
<code_context>
deviceId = payload['commands'][0]['ids'][0]
params = payload['commands'][0]['states']
- mqtt.publish(topic=str(deviceId) + '/' + 'notification',
+ # Multi-tenant MQTT topic: {user_id}/{device_id}/notification
+ mqtt_topic = f"{user_id}/{deviceId}/notification"
+ mqtt.publish(topic=mqtt_topic,
payload=str(params), qos=0) # SENDING MQTT MESSAGE
</code_context>
<issue_to_address>
**🚨 issue (security):** Normalize or sanitize user_id before embedding it in the MQTT topic to avoid malformed topics and mis-attribution.
Because `user_id` can reach this code via `_resolve_smarthome_user_scope` without `_normalize_user_scope`, a client-controlled `agentUserId` with slashes (e.g. `foo/bar`) will produce an MQTT topic like `foo/bar/<deviceId>/notification`. `handle_messages` will then parse this as `user_id='foo'`, `device_id='bar'`, `msg_type='<deviceId>'`, misrouting messages across tenants. Please ensure `user_id` is normalized/sanitized for MQTT topics (e.g. via `_normalize_user_scope` or by enforcing a single topic segment) and fail early on invalid values.
</issue_to_address>
### Comment 2
<location path="notifications.py" line_range="106-117" />
<code_context>
+ # Try to update Firebase if it's a status message
+ if msg_type == 'status':
+ try:
+ from firebase_admin import db
+ # Update status in /users/{user_id}/devices/{device_id}/states
+ # Assuming payload is a JSON string of states
+ import json
+ state_updates = json.loads(payload)
+ db.reference(f'/users/{user_id}/devices/{device_id}/states').update(state_updates)
+ logger.debug("Updated Firebase status for %s/%s", user_id, device_id)
</code_context>
<issue_to_address>
**suggestion (bug_risk):** Guard JSON decoding to avoid noisy errors when payloads are already decoded or non-JSON.
Since `payload` comes from `_decode_payload`, it may already be a Python object or a non-JSON string. Unconditionally calling `json.loads(payload)` will raise for non-JSON payloads, producing noisy "failed to update Firebase" logs and blocking legitimate updates. Consider only calling `json.loads` when `payload` is a `str`, and skipping the Firebase update when JSON decoding fails so expected non-JSON messages don’t look like errors.
```suggestion
# Try to update Firebase if it's a status message
if msg_type == 'status':
# Guard JSON decoding to avoid noisy errors when payloads are already decoded or non-JSON
state_updates = None
if isinstance(payload, dict):
# Payload already decoded by _decode_payload
state_updates = payload
elif isinstance(payload, str):
# Only attempt JSON decoding for string payloads
try:
import json
state_updates = json.loads(payload)
except (ValueError, TypeError):
# Non-JSON string payloads are expected for some messages; skip Firebase update quietly
logger.debug(
"Skipping Firebase status update for non-JSON payload on %s/%s",
user_id,
device_id,
)
else:
# Non-dict, non-str payloads (e.g. bytes) are skipped to avoid noisy errors
logger.debug(
"Skipping Firebase status update for unsupported payload type %s on %s/%s",
type(payload).__name__,
user_id,
device_id,
)
if state_updates is not None:
try:
from firebase_admin import db
# Update status in /users/{user_id}/devices/{device_id}/states
db.reference(f'/users/{user_id}/devices/{device_id}/states').update(state_updates)
logger.debug("Updated Firebase status for %s/%s", user_id, device_id)
except Exception as e:
logger.error("Failed to update Firebase from MQTT: %s", e)
```
</issue_to_address>
### Comment 3
<location path="tests/test_app.py" line_range="452-461" />
<code_context>
+ def test_smarthome_uses_agent_user_scope(self):
</code_context>
<issue_to_address>
**nitpick (testing):** Clarify and strengthen assertions around user scope vs agentUserId precedence in smarthome tests
The test correctly verifies that `_resolve_smarthome_user_scope` prefers request-level `agentUserId` over `current_token.user.id`, but the first inline comment is misleading. Please update it to state that `agentUserId` takes precedence and `current_token.user.id` is a fallback.
Also consider adding a complementary test where `agentUserId` is omitted from the payload, to assert that both `user_id` and `agent_user_id` then come from `current_token.user.id`. This will exercise both the primary and fallback resolution paths.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
Comment on lines
+452
to
+461
| def test_smarthome_uses_agent_user_scope(self): | ||
| payload = { | ||
| 'requestId': '1', | ||
| 'agentUserId': '42', | ||
| 'inputs': [{'intent': 'action.devices.SYNC', 'payload': {}}], | ||
| } | ||
|
|
||
| mock_token = MagicMock() | ||
| mock_token.user.id = 1 | ||
|
|
Contributor
There was a problem hiding this comment.
nitpick (testing): Clarify and strengthen assertions around user scope vs agentUserId precedence in smarthome tests
The test correctly verifies that _resolve_smarthome_user_scope prefers request-level agentUserId over current_token.user.id, but the first inline comment is misleading. Please update it to state that agentUserId takes precedence and current_token.user.id is a fallback.
Also consider adding a complementary test where agentUserId is omitted from the payload, to assert that both user_id and agent_user_id then come from current_token.user.id. This will exercise both the primary and fallback resolution paths.
DaTiC0
added a commit
that referenced
this pull request
May 4, 2026
…se paths, and harden MQTT handling
Owner
Author
|
I have addressed the feedback from the code reviews with the following fixes: 🔐 Security & Consistency
🛠️ Robustness
🧪 Testing
All 43 tests are passing ( 🤖 Resolved via Gemini CLI |
…se paths, and harden MQTT handling
Agent-Logs-Url: https://github.com/DaTiC0/smart-google/sessions/9492fb32-00a8-4614-9b19-f23afbe1a545 Co-authored-by: DaTiC0 <13198638+DaTiC0@users.noreply.github.com>
Agent-Logs-Url: https://github.com/DaTiC0/smart-google/sessions/a7c83973-d307-4a35-8591-9afa75f5f1b1 Co-authored-by: DaTiC0 <13198638+DaTiC0@users.noreply.github.com>
DaTiC0
force-pushed
the
feat/mqtt-connectivity
branch
from
May 5, 2026 05:48
469452f to
ed6a7a4
Compare
…istency and linting compliance
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
This PR introduces a significant architectural shift to support multi-tenancy and enhanced security:
user_id, ensuring message isolation between users.Key Changes
action_devices.pyto enforceuser_idscoping for all SYNC and EXECUTE operations.routes.pyto handleagentUserIdandcurrent_tokenfor consistent user resolution.notifications.pywith thread-safe user-scoped log buffers.tests/test_app.py.tests/test_multi_tenant.py.Test plan
make test).Production Note
While this PR significantly improves the foundation, a follow-up "Production Hardening" phase is recommended to address:
🤖 Generated with Gemini CLI
Summary by Sourcery
Introduce user-scoped device management and MQTT handling to support multi-tenancy and stricter per-user isolation across devices, logs, and smarthome requests.
New Features:
Bug Fixes:
Enhancements:
Documentation:
Tests: