PM-pinou
8fa36b774e
Key changes: - New: data_loader.py SQLite persistence with drop_sqlite_table - New: db_utils.py retry_on_lock decorator (3 retries, exponential backoff) - New: tool_registry.py with 27 MCP tools (filter_and_cluster, compute_scores, etc.) - New: tls_ref.py for TLS cipher/reference data - New: import_tlsdb.py management command - New: scripts/start_server.py for portable runtime - New: migrations 0003-0007 for SQLite table, display_id, llm fields - Changed: views.py unified pipeline worker, retry_run, display_id everywhere - Changed: models.py with display_id auto-assignment, run_type, sqlite_table, llm_thinking, tool_calls_json - Changed: urls.py added retry_run route - Changed: session_store.py robust JSON persistence - Changed: AGENTS.md v7 fix summary added - Changed: templates — globe rewrite (inertia/polar flip), auto.html (thinking/tool accordions), base.html (toast/config), all pages use display_id - Changed: run.bat PYTHONUTF8=1 - Deleted: entity_detector.py, entity_aggregator.py (replaced by filter_and_cluster clustering pipeline) - Test: 92/92 unit tests passing
95 lines
2.9 KiB
Python
95 lines
2.9 KiB
Python
"""SQLite-backed TLS reference data access — replaces ``TlsDB.csv``.
|
|
|
|
Provides functions to query the ``tls_ref_data`` table (populated by
|
|
the ``import_tlsdb`` management command) instead of reading the
|
|
now-deprecated ``TlsDB.csv`` file.
|
|
|
|
Usage::
|
|
|
|
from analysis.tls_ref import get_tls_ref_data, get_tls_field, get_tls_ref_dict
|
|
|
|
# All rows as list of dicts
|
|
all_fields = get_tls_ref_data()
|
|
|
|
# Single field
|
|
field = get_tls_field('0ver')
|
|
# -> {'field_code': '0ver', 'meaning': 'TLS版本', 'data_type': '2字节'}
|
|
|
|
# Dict lookup (cached in memory after first call)
|
|
lookup = get_tls_ref_dict()
|
|
# -> {'0ver': {'meaning': 'TLS版本', 'data_type': '2字节'}, ...}
|
|
"""
|
|
|
|
from functools import lru_cache
|
|
from typing import Optional
|
|
|
|
from django.db import connection
|
|
|
|
|
|
def get_tls_ref_data() -> list[dict[str, str]]:
|
|
"""Return all rows from ``tls_ref_data`` as a list of dicts.
|
|
|
|
Returns:
|
|
Each dict has keys ``field_code``, ``meaning``, ``data_type``.
|
|
Returns an empty list if the table does not exist (e.g. before
|
|
migration).
|
|
"""
|
|
if not _table_exists():
|
|
return []
|
|
|
|
with connection.cursor() as cursor:
|
|
cursor.execute(
|
|
'SELECT field_code, meaning, data_type FROM tls_ref_data ORDER BY id'
|
|
)
|
|
columns = [col[0] for col in cursor.description]
|
|
return [dict(zip(columns, row)) for row in cursor.fetchall()]
|
|
|
|
|
|
def get_tls_field(field_code: str) -> Optional[dict[str, str]]:
|
|
"""Return a single TLS field definition by its code.
|
|
|
|
Args:
|
|
field_code: e.g. ``'0ver'``, ``':ips'``, ``'cipher-suite'``.
|
|
|
|
Returns:
|
|
Dict with keys ``field_code``, ``meaning``, ``data_type``, or
|
|
``None`` if not found.
|
|
"""
|
|
if not _table_exists():
|
|
return None
|
|
|
|
with connection.cursor() as cursor:
|
|
cursor.execute(
|
|
'SELECT field_code, meaning, data_type FROM tls_ref_data WHERE field_code = %s',
|
|
[field_code],
|
|
)
|
|
row = cursor.fetchone()
|
|
if row is None:
|
|
return None
|
|
return {
|
|
'field_code': row[0],
|
|
'meaning': row[1],
|
|
'data_type': row[2],
|
|
}
|
|
|
|
|
|
@lru_cache(maxsize=1)
|
|
def get_tls_ref_dict() -> dict[str, dict[str, str]]:
|
|
"""Return a dict mapping ``field_code -> {meaning, data_type}``.
|
|
|
|
The result is cached in memory after the first call. Call
|
|
``get_tls_ref_dict.cache_clear()`` to invalidate if the
|
|
underlying table is modified at runtime.
|
|
"""
|
|
rows = get_tls_ref_data()
|
|
return {r['field_code']: {'meaning': r['meaning'], 'data_type': r['data_type']} for r in rows}
|
|
|
|
|
|
def _table_exists() -> bool:
|
|
"""Check if the ``tls_ref_data`` table exists in the database."""
|
|
with connection.cursor() as cursor:
|
|
cursor.execute(
|
|
"SELECT name FROM sqlite_master WHERE type='table' AND name='tls_ref_data'"
|
|
)
|
|
return cursor.fetchone() is not None
|