feat(unstable): add consumer/sink class, and demo with a FileConsumer+Plotly sink. - #492
feat(unstable): add consumer/sink class, and demo with a FileConsumer+Plotly sink. #492arsenovic wants to merge 7 commits into
Conversation
 The PR is not safe to merge because valid publisher JSON cannot be consumed, sink shutdown can hang, minimal installations cannot import the sinks package, and the demo leaks its simulator process. Findings
Prompt To Fix All With AI### Issue 1
packages/instro-unstable/instro/unstable/lib/consumers/files.py:47-49
`FileConsumer` claims compatibility with `FilePublisher`, but this branch parses only one line at a time. `FilePublisher(format="json")` writes a single indented JSON array, so the first read contains only `"["` and raises `JSONDecodeError`. As a result, the consumer cannot read valid JSON output produced by the publisher. Parse the complete document and iterate through its records, or remove the unsupported format.
### Issue 2
packages/instro-unstable/instro/unstable/lib/sinks/sink.py:25-29
When this sink uses `FileConsumer` and the file is idle at EOF, the consumer sleeps and retries without yielding. `_consume_loop` checks `_running` only after an item is yielded, so the worker never sees the stop request and this unbounded `join()` never returns. The consumer needs a cancellation path, or shutdown must not depend on another record arriving.
### Issue 3
packages/instro-unstable/instro/unstable/lib/sinks/__init__.py:1
Importing `instro.unstable.lib.sinks`, even just to use the base `Sink`, eagerly imports `PlotlyLiveSink`. That requires NumPy, Plotly, and IPython, but `instro-unstable` does not declare any of them. A minimal supported installation therefore raises `ModuleNotFoundError` when importing this package. Declare these dependencies or put the Plotly integration behind an optional, guarded import.
### Issue 4
packages/instro-unstable/instro/unstable/lib/examples/plotly_sink/simulate_data.py:10
The demo starts a simulator child process but never terminates or waits for it. Because the server runs indefinitely, both normal completion and exceptions leave it running and holding port 5026. Later demo runs can then fail because the port is already in use. Ensure cleanup terminates and waits for the child, including when an exception occurs.
### Issue 5
packages/instro-unstable/instro/unstable/lib/consumers/consumer.py:20-22
The abstract `consume()` contract says it returns a materialized list and describes batch reading, but `FileConsumer` implements an unbounded generator and `Sink` relies on that iterator behavior. This makes the public contract misleading for implementations and type checking because callers cannot tell whether consumption is finite or streaming. Define the abstraction consistently as an iterator or iterable, or separate batch and streaming operations.
Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!
---
For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.
|
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
| yield super().record_to_object(json.loads(line)) | ||
| elif self.format == FileFormats.JSON: | ||
| yield super().record_to_object(json.loads(line)) |
There was a problem hiding this comment.
FileConsumer claims compatibility with FilePublisher, but this branch parses only one line at a time. FilePublisher(format="json") writes a single indented JSON array, so the first read contains only "[" and raises JSONDecodeError. As a result, the consumer cannot read valid JSON output produced by the publisher. Parse the complete document and iterate through its records, or remove the unsupported format.
Prompt To Fix With AI
This is a comment left during a code review.
Path: packages/instro-unstable/instro/unstable/lib/consumers/files.py
Line: 47-49
Comment:
**JSON publisher output fails**
`FileConsumer` claims compatibility with `FilePublisher`, but this branch parses only one line at a time. `FilePublisher(format="json")` writes a single indented JSON array, so the first read contains only `"["` and raises `JSONDecodeError`. As a result, the consumer cannot read valid JSON output produced by the publisher. Parse the complete document and iterate through its records, or remove the unsupported format.
---
For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.| def stop(self): | ||
| """Gracefully stops the background consumption thread.""" | ||
| self._running = False | ||
| if self._thread: | ||
| self._thread.join() |
There was a problem hiding this comment.
When this sink uses FileConsumer and the file is idle at EOF, the consumer sleeps and retries without yielding. _consume_loop checks _running only after an item is yielded, so the worker never sees the stop request and this unbounded join() never returns. The consumer needs a cancellation path, or shutdown must not depend on another record arriving.
Prompt To Fix With AI
This is a comment left during a code review.
Path: packages/instro-unstable/instro/unstable/lib/sinks/sink.py
Line: 25-29
Comment:
**Sink shutdown can hang**
When this sink uses `FileConsumer` and the file is idle at EOF, the consumer sleeps and retries without yielding. `_consume_loop` checks `_running` only after an item is yielded, so the worker never sees the stop request and this unbounded `join()` never returns. The consumer needs a cancellation path, or shutdown must not depend on another record arriving.
---
For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.| @@ -0,0 +1,4 @@ | |||
| from .plotly import PlotlyLiveSink | |||
There was a problem hiding this comment.
Importing instro.unstable.lib.sinks, even just to use the base Sink, eagerly imports PlotlyLiveSink. That requires NumPy, Plotly, and IPython, but instro-unstable does not declare any of them. A minimal supported installation therefore raises ModuleNotFoundError when importing this package. Declare these dependencies or put the Plotly integration behind an optional, guarded import.
Prompt To Fix With AI
This is a comment left during a code review.
Path: packages/instro-unstable/instro/unstable/lib/sinks/__init__.py
Line: 1
Comment:
**Sink imports require undeclared packages**
Importing `instro.unstable.lib.sinks`, even just to use the base `Sink`, eagerly imports `PlotlyLiveSink`. That requires NumPy, Plotly, and IPython, but `instro-unstable` does not declare any of them. A minimal supported installation therefore raises `ModuleNotFoundError` when importing this package. Declare these dependencies or put the Plotly integration behind an optional, guarded import.
---
For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.|
|
||
| # Pass your terminal command as a list of strings | ||
| cmd = ["python", "-m", "instro.dmm.scpi_sim_server", "--dc-voltage", "1", "--dc-current", "1", "--ac-current", "1.02"] | ||
| server = subprocess.Popen(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) |
There was a problem hiding this comment.
The demo starts a simulator child process but never terminates or waits for it. Because the server runs indefinitely, both normal completion and exceptions leave it running and holding port 5026. Later demo runs can then fail because the port is already in use. Ensure cleanup terminates and waits for the child, including when an exception occurs.
Prompt To Fix With AI
This is a comment left during a code review.
Path: packages/instro-unstable/instro/unstable/lib/examples/plotly_sink/simulate_data.py
Line: 10
Comment:
**Simulator process leaks**
The demo starts a simulator child process but never terminates or waits for it. Because the server runs indefinitely, both normal completion and exceptions leave it running and holding port 5026. Later demo runs can then fail because the port is already in use. Ensure cleanup terminates and waits for the child, including when an exception occurs.
---
For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.| def consume(self) -> list[Measurement | Command]: | ||
| """Read all records from the source. This is a convenience method for consumers that support batch reading.""" | ||
| raise NotImplementedError("consume() is not implemented") |
There was a problem hiding this comment.
The abstract consume() contract says it returns a materialized list and describes batch reading, but FileConsumer implements an unbounded generator and Sink relies on that iterator behavior. This makes the public contract misleading for implementations and type checking because callers cannot tell whether consumption is finite or streaming. Define the abstraction consistently as an iterator or iterable, or separate batch and streaming operations.
Prompt To Fix With AI
This is a comment left during a code review.
Path: packages/instro-unstable/instro/unstable/lib/consumers/consumer.py
Line: 20-22
Comment:
**Consumer contract contradicts streaming**
The abstract `consume()` contract says it returns a materialized list and describes batch reading, but `FileConsumer` implements an unbounded generator and `Sink` relies on that iterator behavior. This makes the public contract misleading for implementations and type checking because callers cannot tell whether consumption is finite or streaming. Define the abstraction consistently as an iterator or iterable, or separate batch and streaming operations.
---
For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!
![semgrep-code-nominal-io[bot]](https://avatars.githubusercontent.com/u/116665555?s=60&v=4){kind=small}
| conn.execute( | ||
| f""" | ||
| CREATE TABLE IF NOT EXISTS {self.table_name} ( | ||
| id INTEGER PRIMARY KEY AUTOINCREMENT, | ||
| record_type TEXT NOT NULL, | ||
| payload TEXT NOT NULL, | ||
| created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP | ||
| ) | ||
| """ | ||
| ) |
There was a problem hiding this comment.
Semgrep identified an issue in your code:
The SQLite table name is inserted into a CREATE TABLE statement with an f-string. If an attacker controls table_name, they can alter the schema or prevent the sink from starting.
More details about this
self.table_name is interpolated directly into the SQLite CREATE TABLE statement. Because table_name is supplied to SQLiteSink.__init__, an attacker who can influence that value can change the SQL structure instead of supplying a normal table identifier.
For example:
- An attacker changes configuration so
table_nameisrecords (id INTEGER, attacker_data TEXT) --. _connect()formats that value into the f-string and executesCREATE TABLE IF NOT EXISTS records (id INTEGER, attacker_data TEXT) -- (...).- SQLite creates a schema chosen by the attacker; similar crafted values can cause startup failures or interfere with the sink’s expected table layout.
This is specifically unsafe because table_name is an SQL identifier embedded in the statement, not data handled by the query safely.
To resolve this comment:
✨ Commit fix suggestion
| conn.execute( | |
| f""" | |
| CREATE TABLE IF NOT EXISTS {self.table_name} ( | |
| id INTEGER PRIMARY KEY AUTOINCREMENT, | |
| record_type TEXT NOT NULL, | |
| payload TEXT NOT NULL, | |
| created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP | |
| ) | |
| """ | |
| ) | |
| table_name = self.table_name | |
| if ( | |
| not isinstance(table_name, str) | |
| or not table_name | |
| or not ( | |
| ("A" <= table_name[0] <= "Z") | |
| or ("a" <= table_name[0] <= "z") | |
| or table_name[0] == "_" | |
| ) | |
| or any( | |
| not ( | |
| ("A" <= char <= "Z") | |
| or ("a" <= char <= "z") | |
| or ("0" <= char <= "9") | |
| or char == "_" | |
| ) | |
| for char in table_name[1:] | |
| ) | |
| ): | |
| raise ValueError("table_name must be a valid SQLite identifier") | |
| sql = ( | |
| "CREATE TABLE IF NOT EXISTS " | |
| + table_name | |
| + " (\n" | |
| + "\tid INTEGER PRIMARY KEY AUTOINCREMENT,\n" | |
| + "\trecord_type TEXT NOT NULL,\n" | |
| + "\tpayload TEXT NOT NULL,\n" | |
| + "\tcreated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP\n" | |
| + ")" | |
| ) | |
| conn.execute(sql) |
View step-by-step instructions
-
Decide whether custom table names are required. If not, remove
table_namefrom the constructor and use the fixed identifierrecordsdirectly in theCREATE TABLEstatement. -
If custom names are required, validate
self.table_namebefore building the SQL. Accept only a SQLite identifier matching^[A-Za-z_][A-Za-z0-9_]*$; raiseValueErrorfor any other value. -
Remove the f-string from
conn.execute. Build the statement by concatenating the validated table name between static SQL fragments, for examplesql = "CREATE TABLE IF NOT EXISTS " + self.table_name + " (...)", then callconn.execute(sql). -
Keep parameter binding for values such as
record_typeandpayloadin insert statements, using placeholders like?and passing the values separately. SQL parameters cannot be used for table names, so strict identifier validation is required before inserting the name into the DDL.
💬 Ignore this finding
Reply with Semgrep commands to ignore this finding.
/fp <comment>for false positive/ar <comment>for acceptable risk/other <comment>for all other reasons
Alternatively, triage in Semgrep AppSec Platform to ignore the finding created by formatted-sql-query.
You can view more details about this finding in the Semgrep AppSec Platform.
| self._conn.execute( | ||
| f"INSERT INTO {self.table_name} (record_type, payload) VALUES (?, ?)", | ||
| (record_type, payload), | ||
| ) |
There was a problem hiding this comment.
Semgrep identified a blocking 🔴 issue in your code:
Untrusted self.table_name is embedded in the SQL f-string, allowing an attacker who controls it to rewrite the INSERT query and extract database data.
More details about this
record_type and payload are passed as SQLite parameters, but self.table_name is inserted directly into the SQL with an f-string. If an attacker can influence table_name through configuration or another input path, they can change the INSERT structure instead of merely choosing a table. For example, setting it to audit (record_type, payload) SELECT username, password FROM users WHERE ? IS NOT NULL OR ? IS NULL -- produces a query that copies users.username and users.password into audit; the two ? markers consume the existing record_type and payload arguments, and -- comments out the original VALUES clause. Calling process_item(...) then executes that attacker-supplied query while holding self._lock. The JSON payload itself is not the injection point because it is bound separately; the unsafe interpolation is the table identifier.
Summary: self.table_name is interpolated into an SQLite INSERT, allowing a value controlled by an attacker to rewrite the query and potentially copy sensitive database data.
To resolve this comment:
✨ Commit fix suggestion
| self._conn.execute( | |
| f"INSERT INTO {self.table_name} (record_type, payload) VALUES (?, ?)", | |
| (record_type, payload), | |
| ) | |
| insert_sql = { | |
| "records": "INSERT INTO records (record_type, payload) VALUES (?, ?)", | |
| "archive": "INSERT INTO archive (record_type, payload) VALUES (?, ?)", | |
| }[self.table_name] | |
| self._conn.execute(insert_sql, (record_type, payload)) |
View step-by-step instructions
-
Remove the f-string from the SQL statement. Keep
record_typeandpayloadas bound parameters because they are already passed separately with?placeholders. -
Use a fixed SQL statement when the sink writes to one known table:
self._conn.execute("INSERT INTO records (record_type, payload) VALUES (?, ?)", (record_type, payload)) -
If
self.table_namemust be configurable, restrict it to a fixed allowlist and select a complete static statement instead of interpolating the identifier:
insert_sql = {"records": "INSERT INTO records (record_type, payload) VALUES (?, ?)", "archive": "INSERT INTO archive (record_type, payload) VALUES (?, ?)"}[self.table_name]Then execute
self._conn.execute(insert_sql, (record_type, payload)). Reject any table name that is not in the allowlist. -
Do not try to bind the table name as
?; SQL parameters can represent values, but not table or column identifiers.
💬 Ignore this finding
Reply with Semgrep commands to ignore this finding.
/fp <comment>for false positive/ar <comment>for acceptable risk/other <comment>for all other reasons
Alternatively, triage in Semgrep AppSec Platform to ignore the finding created by sqlalchemy-execute-raw-query.
You can view more details about this finding in the Semgrep AppSec Platform.
| conn.execute( | ||
| f""" | ||
| CREATE TABLE IF NOT EXISTS {self.table_name} ( | ||
| id INTEGER PRIMARY KEY AUTOINCREMENT, | ||
| record_type TEXT NOT NULL, | ||
| payload TEXT NOT NULL, | ||
| created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP | ||
| ) | ||
| """ | ||
| ) |
There was a problem hiding this comment.
Semgrep identified a blocking 🔴 issue in your code:
Untrusted self.table_name is interpolated into SQLite DDL, allowing an attacker controlling sink configuration to alter the created schema or make _connect() fail. The issue affects table setup, not the serialized record values.
More details about this
self.table_name is interpolated directly into the SQLite CREATE TABLE statement. Although the default value (records) is trusted, any caller that can set table_name controls part of the SQL grammar rather than just a table identifier.
For example, an attacker who can influence the sink configuration could supply:
records (attacker_value TEXT PRIMARY KEY) WITHOUT ROWID --
_connect() would then execute SQL equivalent to:
CREATE TABLE IF NOT EXISTS records
(attacker_value TEXT PRIMARY KEY) WITHOUT ROWID -- (...)The injected text changes the schema created by _connect(), and a crafted value can instead make the statement invalid, preventing start() from establishing the sink and causing a denial of service. The payload is not inserted through _serialize_item(); the risk occurs earlier, when self.table_name is assembled into the DDL string. SQLite’s execute() generally rejects multiple statements, so a payload such as records; DROP TABLE users;-- will typically fail rather than run both statements, but it still demonstrates that untrusted configuration reaches SQL syntax directly.
To resolve this comment:
✨ Commit fix suggestion
| conn.execute( | |
| f""" | |
| CREATE TABLE IF NOT EXISTS {self.table_name} ( | |
| id INTEGER PRIMARY KEY AUTOINCREMENT, | |
| record_type TEXT NOT NULL, | |
| payload TEXT NOT NULL, | |
| created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP | |
| ) | |
| """ | |
| ) | |
| conn.execute( | |
| """ | |
| CREATE TABLE IF NOT EXISTS records ( | |
| id INTEGER PRIMARY KEY AUTOINCREMENT, | |
| record_type TEXT NOT NULL, | |
| payload TEXT NOT NULL, | |
| created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP | |
| ) | |
| """ | |
| ) |
View step-by-step instructions
- Remove
self.table_namefrom the SQL statement. SQL parameters can safely bind values, but they cannot bind table names or other SQL identifiers. - Use a fixed table name and execute a static SQL string, for example:
conn.execute("""CREATE TABLE IF NOT EXISTS records (...)"""). - Remove or restrict the
table_nameconstructor argument so callers cannot select an arbitrary database identifier. - Alternatively, if multiple table names are required, map each permitted name to a complete static SQL statement, such as
{"records": "CREATE TABLE ...", "events": "CREATE TABLE ..."}, and execute only a statement selected from that allowlist. Do not build SQL with an f-string, concatenation,%, or.format().
💬 Ignore this finding
Reply with Semgrep commands to ignore this finding.
/fp <comment>for false positive/ar <comment>for acceptable risk/other <comment>for all other reasons
Alternatively, triage in Semgrep AppSec Platform to ignore the finding created by sqlalchemy-execute-raw-query.
You can view more details about this finding in the Semgrep AppSec Platform.
1 participant
Summary
This adds two new core objects :
ConsumerandSink.it demonstrates how to use a FileConsumer and a PlotlyLiveSink (in a notebook) with a simulated DMM as a datasource.
Not ready to merge, just opening for initial review.
Type of change
fix)feat)feat!/fix!)refactor)docs)chore)Verification
How did you verify this change works? Provide evidence the reviewer can evaluate without reproducing your setup — e.g. test output, screenshots, logs, repro steps + result. If the change is hardware-specific and you don't have the device, say so.
Tests
Checklist
feat(driver): add support for Keysight E36300)Notes for reviewers
Optional — call out specific files, edge cases, or decisions you'd like eyes on.