Skip to content

Upgrade Flask app to Python 3.12: Authlib, Flask 3, SQLAlchemy 2, factory pattern, security hardening #98

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
Copilot wants to merge 3 commits into
base: main
Choose a base branch
from
Draft

Upgrade Flask app to Python 3.12: Authlib, Flask 3, SQLAlchemy 2, factory pattern, security hardening #98

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
d07b94c
Initial plan
Copilot Mar 8, 2026
ebdeb7c
Upgrade to Python 3.12: Authlib, Flask 3, SQLAlchemy 2, factory patte…
Copilot Mar 8, 2026
9fc1b7f
Fix action_devices.py: remove duplicated legacy code left after refactor
Copilot Mar 8, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 9 additions & 5 deletions ReportState.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,9 +1,12 @@
import logging
import time
import json
from flask import current_app
from google.auth import crypt, jwt
import requests

logger = logging.getLogger(__name__)


def generate_jwt(service_account):
signer = crypt.RSASigner.from_string(service_account['private_key'])
Expand All @@ -16,6 +19,7 @@ def generate_jwt(service_account):
'scope': 'https://www.googleapis.com/auth/homegraph'
}

# google-auth >= 2.x returns a str; no .decode() needed
return jwt.encode(signer, payload)


Expand All @@ -41,19 +45,19 @@ def report_state(access_token, report_state_file):
}
data = report_state_file
response = requests.post(url, headers=headers, json=data)
print('Response: ' + response.text)
logger.info('Response: %s', response.text)

return response.status_code == requests.codes.ok


def main(report_state_file):
service_account = current_app.config['SERVICE_ACCOUNT_DATA']
print('By ReportState')
signed_jwt = generate_jwt(service_account).decode("utf-8") # Decode
logger.info('By ReportState')
signed_jwt = generate_jwt(service_account)
access_token = get_access_token(signed_jwt)
success = report_state(access_token, report_state_file)

if success:
print('Report State has been done successfully.')
logger.info('Report State has been done successfully.')
else:
print('Report State failed. Please check the log above.')
logger.error('Report State failed. Please check the log above.')
144 changes: 80 additions & 64 deletions action_devices.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,25 +2,28 @@
# Code By DaTi_Co

import json
import logging
import requests
from flask import current_app
from notifications import mqtt

logger = logging.getLogger(__name__)

try:
import ReportState as state
REPORTSTATE_AVAILABLE = True
except ImportError:
state = None
REPORTSTATE_AVAILABLE = False
print("ReportState module not available, some features may be disabled.")
logger.warning("ReportState module not available, some features may be disabled.")

# Try to import firebase_admin, but provide fallback if not available
try:
from firebase_admin import db
FIREBASE_AVAILABLE = True
except ImportError:
FIREBASE_AVAILABLE = False
print("Firebase admin not available, using mock data for testing")
logger.warning("Firebase admin not available, using mock data for testing")

# Mock data for testing when Firebase is not available
MOCK_DEVICES = {
Expand Down Expand Up @@ -90,7 +93,7 @@
return db.reference('/devices')
except Exception as e:
# Firebase is installed but not initialized (e.g. missing credentials in dev)
print(f"Firebase not initialized, falling back to mock data: {e}")
logger.warning("Firebase not initialized, falling back to mock data: %s", e)
return MockRef()


Expand All @@ -109,15 +112,15 @@
}
for device in devices:
device = str(device)
print('\nGetting Device status from: ' + device)
logger.debug('Getting Device status from: %s', device)
state_data = rquery(device)
if state_data:
payload['devices']['states'][device] = state_data
print(state_data)
logger.debug('Device state: %s', state_data)

return payload
except Exception as e:
print(f"Error in rstate: {e}")
logger.error("Error in rstate: %s", e)
return {"devices": {"states": {}}}


Expand All @@ -139,7 +142,7 @@
DEVICES.append(DEVICE)
return DEVICES
except Exception as e:
print(f"Error in rsync: {e}")
logger.error("Error in rsync: %s", e)
return []


Expand All @@ -148,7 +151,7 @@
ref = reference()
return ref.child(deviceId).child('states').get()
except Exception as e:
print(f"Error querying device {deviceId}: {e}")
logger.error("Error querying device %s: %s", deviceId, e)
return {"online": False}


Expand All @@ -158,7 +161,7 @@
ref.child(deviceId).child('states').update(parameters)
return ref.child(deviceId).child('states').get()
except Exception as e:
print(f"Error executing on device {deviceId}: {e}")
logger.error("Error executing on device %s: %s", deviceId, e)
return parameters


Expand All @@ -169,7 +172,7 @@
"devices": rsync()
}
except Exception as e:
print(f"Error in onSync: {e}")
logger.error("Error in onSync: %s", e)
return {"agentUserId": "test-user", "devices": []}


Expand All @@ -182,12 +185,12 @@
for i in body['inputs']:
for device in i['payload']['devices']:
deviceId = device['id']
print('DEVICE ID: ' + deviceId)
logger.debug('DEVICE ID: %s', deviceId)
data = rquery(deviceId)
payload['devices'][deviceId] = data
return payload
except Exception as e:
print(f"Error in onQuery: {e}")
logger.error("Error in onQuery: %s", e)
return {"devices": {}}


Expand All @@ -211,81 +214,93 @@
for execution in command['execution']:
execCommand = execution['command']
params = execution['params']
# First try to refactor
payload = commands(payload, deviceId, execCommand, params)
return payload
except Exception as e:
print(f"Error in onExecute: {e}")
logger.error("Error in onExecute: %s", e)
return {'commands': [{'ids': [], 'status': 'ERROR', 'errorCode': 'deviceNotFound'}]}


def commands(payload, deviceId, execCommand, params):
""" more clean code as was bedore.
dont remember how state ad parameters is used """
"""Map an execution command to its device-state parameters and apply them."""
# Dispatch map: command → parameter transformer
_COMMAND_PARAMS = {
'action.devices.commands.OnOff': lambda p: {'on': p['on']} if 'on' in p else None,
'action.devices.commands.BrightnessAbsolute': lambda p: {'brightness': p.get('brightness', 100), 'on': True},
'action.devices.commands.StartStop': lambda p: {'isRunning': p['start']},
'action.devices.commands.PauseUnpause': lambda p: {'isPaused': p['pause']},
'action.devices.commands.GetCameraStream': lambda p: p,
'action.devices.commands.LockUnlock': lambda p: {'isLocked': p['lock']},
}

try:
if execCommand == 'action.devices.commands.OnOff':
if 'on' not in params:
print("Error: 'on' parameter missing for OnOff command")
transformer = _COMMAND_PARAMS.get(execCommand)
if transformer is None:
logger.debug('Unhandled command: %s', execCommand)
else:
transformed = transformer(params)
if transformed is None:
logger.error("'on' parameter missing for OnOff command")
payload['commands'][0]['status'] = 'ERROR'
payload['commands'][0]['errorCode'] = 'hardError'
return payload
params = {'on': params['on']}
print('OnOff')
elif execCommand == 'action.devices.commands.BrightnessAbsolute':
params = {'brightness': params.get('brightness', 100), 'on': True}
print('BrightnessAbsolute')
elif execCommand == 'action.devices.commands.StartStop':
params = {'isRunning': params['start']}
print('StartStop')
elif execCommand == 'action.devices.commands.PauseUnpause':
params = {'isPaused': params['pause']}
print('PauseUnpause')
elif execCommand == 'action.devices.commands.GetCameraStream':
print('GetCameraStream')
elif execCommand == 'action.devices.commands.LockUnlock':
params = {'isLocked': params['lock']}
print('LockUnlock')

# Out from elif
params = transformed
logger.debug('Executing command: %s', execCommand)

states = rexecute(deviceId, params)
payload['commands'][0]['states'] = states

return payload
except Exception as e:
print(f"Error in commands: {e}")
logger.error("Error in commands: %s", e)
payload['commands'][0]['status'] = 'ERROR'
return payload


def _handle_execute(req):
"""Execute intent handler – runs onExecute and publishes MQTT notification."""
payload = onExecute(req)
try:
if (payload.get('commands')
and payload['commands'][0]['ids']):
deviceId = payload['commands'][0]['ids'][0]
params = payload['commands'][0]['states']
mqtt.publish(
topic=str(deviceId) + '/notification',
payload=str(params),
qos=0,
)
except Exception as mqtt_error:
logger.warning("MQTT error: %s", mqtt_error)
return payload


# ---------------------------------------------------------------------------
# Dispatch map: Google Home intent → handler function
# ---------------------------------------------------------------------------
_INTENT_DISPATCH = {
"action.devices.SYNC": lambda req: onSync(),
"action.devices.QUERY": onQuery,
"action.devices.EXECUTE": _handle_execute,
"action.devices.DISCONNECT": lambda req: {},
}


def actions(req):
try:
payload = {}
for i in req['inputs']:
print(i['intent'])
if i['intent'] == "action.devices.SYNC":
payload = onSync()
elif i['intent'] == "action.devices.QUERY":
payload = onQuery(req)
elif i['intent'] == "action.devices.EXECUTE":
payload = onExecute(req)
# SEND TEST MQTT
try:
if payload.get('commands') and len(payload['commands']) > 0 and len(payload['commands'][0]['ids']) > 0:
deviceId = payload['commands'][0]['ids'][0]
params = payload['commands'][0]['states']
mqtt.publish(topic=str(deviceId) + '/' + 'notification',
payload=str(params), qos=0) # SENDING MQTT MESSAGE
except Exception as mqtt_error:
print(f"MQTT error: {mqtt_error}")
elif i['intent'] == "action.devices.DISCONNECT":
print("\nDISCONNECT ACTION")
payload = {}
intent = i['intent']
logger.debug('Intent: %s', intent)
handler = _INTENT_DISPATCH.get(intent)
if handler is not None:
payload = handler(req)
else:
print('Unexpected action requested: %s', json.dumps(req))
logger.warning('Unexpected action requested: %s', json.dumps(req))
payload = {}
return payload
except Exception as e:
print(f"Error in actions: {e}")
logger.error("Error in actions: %s", e)
return {}


Expand All @@ -297,19 +312,19 @@

response = requests.post(url, json=data)

print(f'\nRequests Code: {requests.codes["ok"]}\nResponse Code: {response.status_code}')
print(f'\nResponse: {response.text}')
logger.debug('Requests Code: %s Response Code: %s', requests.codes["ok"], response.status_code)
logger.debug('Response: %s', response.text)

Check failure

Code scanning / CodeQL

Clear-text logging of sensitive information High

This expression logs
sensitive data (password)
Loading
as clear text.
This expression logs
sensitive data (password)
Loading
as clear text.
This expression logs
sensitive data (password)
Loading
as clear text.

Copilot Autofix

AI 10 days ago

In general, to fix clear‑text logging of sensitive information you should avoid logging any value that may contain secrets (API keys, tokens, passwords, or responses that could include them). Instead, log only non‑sensitive metadata (e.g., HTTP status code, high‑level result). If detailed body content is ever needed for debugging, it should be gated behind explicit development / debug flags and must be carefully sanitized before logging.

For this specific case, the best minimal fix is to stop logging response.text from the request_sync call, since that response is tainted by the API key used in the request. We already log the response status code on the previous line, which is usually sufficient in production. If keeping some indication of the response is desirable, we can log a short, generic message (e.g., “Response body omitted for security reasons”) instead of the raw text. This preserves existing functionality (the HTTP call and return value are unchanged) while eliminating the sensitive sink.

Concretely:

  • In action_devices.py, inside request_sync, replace logger.debug('Response: %s', response.text) with a non‑sensitive debug line that does not include the body, e.g. logger.debug('Response body omitted from logs for security.') or simply remove that line.
  • No changes are needed in app.py or routes.py beyond how they call request_sync, because the leak occurs when the response body is logged, not when the key is passed into the function.

No new imports or helper methods are required.

Suggested changeset 1
action_devices.py

Autofix patch

Autofix patch
Run the following command in your local git repository to apply this patch
cat << 'EOF' | git apply
diff --git a/action_devices.py b/action_devices.py
--- a/action_devices.py
+++ b/action_devices.py
@@ -313,7 +313,8 @@
         response = requests.post(url, json=data)
 
         logger.debug('Requests Code: %s  Response Code: %s', requests.codes["ok"], response.status_code)
-        logger.debug('Response: %s', response.text)
+        # Do not log response.text directly to avoid leaking sensitive information.
+        logger.debug('Response body omitted from logs for security.')
 
         return response.status_code == requests.codes['ok']
     except Exception as e:
EOF
@@ -313,7 +313,8 @@
response = requests.post(url, json=data)

logger.debug('Requests Code: %s Response Code: %s', requests.codes["ok"], response.status_code)
logger.debug('Response: %s', response.text)
# Do not log response.text directly to avoid leaking sensitive information.
logger.debug('Response body omitted from logs for security.')

return response.status_code == requests.codes['ok']
except Exception as e:
Copilot is powered by AI and may make mistakes. Always verify output.

return response.status_code == requests.codes['ok']
except Exception as e:
print(f"Error in request_sync: {e}")
logger.error("Error in request_sync: %s", e)
return False


def report_state():
try:
if not REPORTSTATE_AVAILABLE:
print("ReportState module not available, skipping report_state")
logger.warning("ReportState module not available, skipping report_state")
return "ReportState not available"
import random
n = random.randint(10**19, 10**20)
Expand All @@ -323,5 +338,6 @@

return "THIS IS TEST NO RETURN"
except Exception as e:
print(f"Error in report_state: {e}")
logger.error("Error in report_state: %s", e)
return f"Error: {e}"

Loading