گزارش رویداد حسگر در پایتون؛ JSON با فیلدهای محدود و آزمون تک‌خطی

گزارش رویداد وقتی قابل بررسی است که شکل و معنای فیلدهایش روشن باشد. به جای نوشتن شیء کامل درخواست، می‌توان رویدادی کوچک با نام ثابت، نتیجه و چند مقدار محدود ساخت. در این درس، دو رویداد ساختگی مربوط به دادهٔ حسگر را با logging و JSON پایتون ثبت و دوباره تجزیه می‌کنیم. مقصد فقط حافظهٔ برنامه است؛ سرور، تماس، فایل، حسگر یا سامانهٔ جمع‌آوری واقعی نداریم و نتیجه، ارزیابی امنیتی محصول نیست.

۱. قرارداد کوچک رویداد را پیش از ثبت بنویسیم

ورودی نمونه دقیقاً سه فیلد دارد: seq برای شماره، status برای کد وضعیت و elapsed_ms برای مدت صحیح بر حسب میلی‌ثانیه. شماره باید مثبت باشد و مدت، صفر یا مثبت. سقف هر دو را ۲۱۴۷۴۸۳۶۴۷ گذاشته‌ایم؛ این حد، انتخاب طرح دادهٔ آموزشی است، نه ظرفیت حسگر، مهلت سرویس یا بهترین سیاست عمومی. مقادیر ۴ و ۹ میلی‌ثانیه در مثال تألیفی‌اند و با ساعت اندازه‌گیری نشده‌اند.

برای وضعیت، تنها دو عضو Enum به نام‌های ACCEPTED و REJECTED پذیرفته می‌شوند. رشتهٔ آزاد، حتی با متن accepted، همان عضو Enum نیست. این وضعیت‌ها نتیجهٔ فرضی یک مرحله‌اند؛ تابع ثبت، خودِ ولتاژ یا کارکرد حسگر را بررسی نمی‌کند. نوع دقیق int برای شماره و مدت الزامی است، بنابراین بولی، عدد اعشاری، NaN و بی‌نهایت وارد قرارداد نمی‌شوند. تبدیل خودکار رشتهٔ عددی یا گرد کردن مدت نداریم.

خروجی چهار کلید ثابت دارد؛ تابع، نام رویداد sensor.validation را اضافه می‌کند و مقدار وضعیت را به کد متنی محدود تبدیل می‌کند. شناسهٔ شماره، رمز نشست یا مجوز دسترسی نیست و یکتایی جهانی آن تضمین نشده است. برای آشنایی با اعتبارسنجی خودِ داده پیش از این مرحله، درس قرارداد دادهٔ حسگر با dataclass را بخوانید؛ این دو قرارداد مسئولیت جدا دارند.

۲. داده را انتخاب کنیم، سپس وارد logging کنیم

تابع event_payload ابتدا شکل dict و مجموعهٔ دقیق کلیدها را بررسی می‌کند. کلید گمشده یا اضافه، حتی payload تو‌در‌تو، استثنا می‌دهد. سپس مقدارهای مجاز را به یک dict تازه منتقل می‌کند؛ از تبدیل عمومی شیء، چاپ همهٔ فیلدها یا ثبت متن ورودی استفاده نمی‌کند. شیء ورودی هم تغییر داده نمی‌شود. این روش، انتخاب فیلد است؛ سازوکار تشخیص خودکار هر نوع اطلاعات حساس در همهٔ قسمت‌های برنامه نیست.

پس از اعتبارسنجی، JSON ساخته و تنها رشتهٔ حاصل به logger داده می‌شود. طبق مستندات logging، logger رکورد می‌سازد و handler آن را به مقصد می‌فرستد. در آزمون، شمار فراخوانی makeRecord همین نمونه را مشاهده می‌کنیم: دو ورودی معتبر، دو رکورد می‌سازند؛ ورودی‌های نامعتبرِ بعدی، رکورد یا خط تازه‌ای ایجاد نمی‌کنند. این بررسی به ترتیب همین مسیر مربوط است، نه تمام مسیرهای گزارش یک برنامهٔ بزرگ.

راهنمای OWASP بر اعتبارسنجی دادهٔ رویداد و رمزگذاری متناسب با قالب تأکید می‌کند. همچنین خطای یک فیلد نباید موجب حذف خاموش کل رخداد امنیتی شود. نمونهٔ ما رابط محدود آموزشی است: نقص قرارداد به فراخواننده استثنا می‌دهد و رویداد امنیتی جایگزین نمی‌سازد. برنامهٔ واقعی باید رسیدگی به این خطا را با زمینهٔ محدود و امن طراحی کند؛ بازنشر همان ورودی خراب راه‌حل مناسبی نیست.

۳. JSON تک‌خطی با logger مستقل در حافظه

با json.dumps، گزینهٔ ensure_ascii را روشن، allow_nan را خاموش و جداکننده‌ها را فشرده انتخاب می‌کنیم. قالب formatter فقط message است و StreamHandler یک پایان خط به آن اضافه می‌کند. بنابراین هر فراخوانی معتبر، یک متن JSON روی یک خط می‌دهد. این مجموعه یک سند JSON آرایه‌ای نیست؛ هنگام خواندن، هر خط را جدا با loads تجزیه می‌کنیم.

برای آزمایش فراردهی، رشتهٔ فارسی دارای خط جدید و بازگشت سطر را جداگانه سریال می‌کنیم. رشتهٔ رمزگذاری‌شده، کنترل خط واقعی ندارد و پس از loads همان متن اولیه برمی‌گردد. این آزمایش از مسیر رویداد برنامه عبور نمی‌کند؛ طرح اصلی اصلاً فیلد متن آزاد ندارد و وضعیت دارای خط جدید را رد می‌کند. فراردهی، محتوا را محرمانه یا مناسب هر مقصد دیگری نمی‌کند؛ فقط رفتار همین قالب را می‌آزماید.

logger نام مخصوص آزمون دارد، یک handler می‌گیرد و propagate آن خاموش است. پیکربندی root یا کارخانهٔ سراسری LogRecord را عوض نمی‌کنیم. شمارنده با patch.object فقط makeRecord همین شیء را موقتاً می‌پوشاند؛ سپس پوشش و handler برداشته می‌شوند. این ابزار مشاهدهٔ آزمون است و لازم نیست بخش ثابت منطق رویداد در برنامهٔ واقعی باشد.

فایل structured_sensor_logs.py را ذخیره و اجرا کنید. اجرای ثبت‌شده Python 3.14.5 دارد؛ نسخهٔ مستندات خوانده‌شده 3.14.8 بوده است. کد کامل، بستن مقصد حافظه‌ای و برداشتن handler را نیز نشان می‌دهد. نام logger در خروجی JSON قرار نمی‌گیرد، بنابراین نام متفاوت محیط آزمون، رکوردهای تألیفی را تغییر نمی‌دهد.

"""Allowlisted synthetic JSON telemetry events in a private in-memory logger only."""
from contextlib import contextmanager
from enum import Enum
import io
import json
import logging
import math
import platform
from unittest.mock import patch
import uuid


MAX_COUNT = 2**31 - 1  # Authored schema bound, not a sensor or timing capability.


class Status(Enum):
    ACCEPTED = 'accepted'
    REJECTED = 'rejected'


def event_payload(fields):
    if type(fields) is not dict:
        raise TypeError('Exact dict event fields required')
    if set(fields) != {'seq', 'status', 'elapsed_ms'}:
        raise ValueError('Exactly seq, status and elapsed_ms are required')
    seq, status, elapsed_ms = fields['seq'], fields['status'], fields['elapsed_ms']
    if type(seq) is not int or type(elapsed_ms) is not int:
        raise TypeError('Exact int required; bool, float and nonfinite values excluded')
    if not 1 <= seq <= MAX_COUNT or not 0 <= elapsed_ms <= MAX_COUNT:
        raise ValueError('Sequence or elapsed milliseconds outside authored bounds')
    if type(status) is not Status:
        raise TypeError('An explicit Status enum member is required')
    return {'event': 'sensor.validation', 'seq': seq,
            'status': status.value, 'elapsed_ms': elapsed_ms}


def emit_event(logger, fields):
    # Validate and serialize before invoking Logger.log / creating a LogRecord.
    payload = event_payload(fields)
    line = json.dumps(payload, ensure_ascii=True, allow_nan=False, separators=(',', ':'))
    level = logging.INFO if fields['status'] is Status.ACCEPTED else logging.WARNING
    logger.log(level, line)
    return line


class RecordingHandler(logging.StreamHandler):
    def __init__(self, stream):
        super().__init__(stream)
        self.level_names = []

    def emit(self, record):
        self.level_names.append(record.levelname)
        super().emit(record)


@contextmanager
def private_sink():
    stream = io.StringIO()
    # A unique fixture-only name avoids reconfiguring an application's logger.
    logger = logging.getLogger('lesson.synthetic.sensor.' + uuid.uuid4().hex)
    original_level, original_propagation = logger.level, logger.propagate
    logger.setLevel(logging.INFO)
    logger.propagate = False
    handler = RecordingHandler(stream)
    handler.setFormatter(logging.Formatter('%(message)s'))
    logger.addHandler(handler)
    try:
        # Wrap only this logger instance, never the global LogRecord factory.
        with patch.object(logger, 'makeRecord', wraps=logger.makeRecord) as creation_probe:
            yield logger, stream, creation_probe, handler
    finally:
        logger.removeHandler(handler)
        handler.close()
        stream.close()
        logger.setLevel(original_level)
        logger.propagate = original_propagation


def rejects(call, exception):
    try:
        call()
    except exception:
        return True
    return False


def run_example():
    root = logging.getLogger()  # Read configuration only.
    root_before = (root.level, tuple(root.handlers), tuple(root.filters))
    factory_before = logging.getLogRecordFactory()
    first = {'seq': 7, 'status': Status.ACCEPTED, 'elapsed_ms': 4}
    second = {'seq': 8, 'status': Status.REJECTED, 'elapsed_ms': 9}
    first_before = dict(first)
    with private_sink() as (logger, stream, creation_probe, handler):
        emitted = [emit_event(logger, first), emit_event(logger, second)]
        lines = stream.getvalue().splitlines()
        decoded = [json.loads(line) for line in lines]
        after_valid = creation_probe.call_count
        checks = {
            'two_calls_produce_exactly_two_physical_lines': len(lines) == 2 and stream.getvalue().count('\n') == 2,
            'json_roundtrip_matches_authored_records': decoded == [
                {'event': 'sensor.validation', 'seq': 7, 'status': 'accepted', 'elapsed_ms': 4},
                {'event': 'sensor.validation', 'seq': 8, 'status': 'rejected', 'elapsed_ms': 9}],
            'each_record_has_only_four_fixed_keys': all(set(row) == {'event', 'seq', 'status', 'elapsed_ms'} for row in decoded),
            'output_numeric_fields_remain_exact_integers': all(type(row['seq']) is int and type(row['elapsed_ms']) is int for row in decoded),
            'status_codes_map_to_authored_info_warning_levels': handler.level_names == ['INFO', 'WARNING'],
            'bool_sequence_rejected': rejects(lambda: emit_event(logger, {**first, 'seq': True}), TypeError),
            'bool_elapsed_rejected': rejects(lambda: emit_event(logger, {**first, 'elapsed_ms': False}), TypeError),
            'out_of_range_sequence_and_elapsed_rejected': all(rejects(lambda field=field, value=value: emit_event(logger, {**first, field: value}), ValueError)
                for field, value in (('seq', 0), ('seq', MAX_COUNT + 1), ('elapsed_ms', -1), ('elapsed_ms', MAX_COUNT + 1))),
            'float_nan_and_infinity_rejected_before_logging': all(rejects(lambda value=value: emit_event(logger, {**first, 'elapsed_ms': value}), TypeError)
                for value in (4.0, math.nan, math.inf, -math.inf)),
            'arbitrary_or_newline_status_text_rejected': all(rejects(lambda value=value: emit_event(logger, {**first, 'status': value}), TypeError)
                for value in ('accepted', 'unknown', 'accepted\nforged', None)),
            'missing_unknown_and_nondict_inputs_rejected': rejects(lambda: emit_event(logger, {'seq': 7, 'status': Status.ACCEPTED}), ValueError)
                and rejects(lambda: emit_event(logger, {**first, 'payload': {'text': 'DO_NOT_LOG'}}), ValueError)
                and rejects(lambda: emit_event(logger, [('seq', 7)]), TypeError),
            'invalid_inputs_create_no_logrecord_or_line': creation_probe.call_count == after_valid == 2 and stream.getvalue().splitlines() == lines,
            'private_logger_has_one_handler_without_propagation': len(logger.handlers) == 1 and logger.propagate is False,
            'input_is_not_mutated_and_serialization_is_repeatable': first == first_before and emitted[0] == json.dumps(event_payload(first), ensure_ascii=True, allow_nan=False, separators=(',', ':')),
            'zero_elapsed_and_inclusive_schema_boundaries_accepted': event_payload({'seq': MAX_COUNT, 'status': Status.ACCEPTED, 'elapsed_ms': 0})['elapsed_ms'] == 0
                and event_payload({'seq': 1, 'status': Status.ACCEPTED, 'elapsed_ms': MAX_COUNT})['elapsed_ms'] == MAX_COUNT,
        }
        levels = list(handler.level_names)
    # This free-text probe is serializer-only, not an allowed application event.
    probe = {'probe': 'حسگر\nسطر دوم\r\n'}
    encoded_probe = json.dumps(probe, ensure_ascii=True, allow_nan=False, separators=(',', ':'))
    checks['separate_json_probe_escapes_controls_and_preserves_unicode'] = (
        '\n' not in encoded_probe and '\r' not in encoded_probe and encoded_probe.isascii()
        and json.loads(encoded_probe) == probe)
    checks['json_nan_encoding_is_explicitly_disabled'] = rejects(lambda: json.dumps({'probe': math.nan}, allow_nan=False), ValueError)
    with private_sink() as (fresh_logger, fresh_stream, fresh_probe, fresh_handler):
        emit_event(fresh_logger, first)
        second_sink_ok = fresh_probe.call_count == 1 and len(fresh_stream.getvalue().splitlines()) == 1
    checks['fresh_sink_does_not_accumulate_handlers_or_change_root_configuration'] = (
        second_sink_ok and root_before == (root.level, tuple(root.handlers), tuple(root.filters))
        and factory_before is logging.getLogRecordFactory())
    assert len(checks) == 18 and all(checks.values()), checks
    return {'python': platform.python_version(), 'scope': 'private_StringIO_synthetic_event_fixture',
            'schema': {'input_keys': ['seq', 'status', 'elapsed_ms'],
                       'output_keys': ['event', 'seq', 'status', 'elapsed_ms'],
                       'event_code': 'sensor.validation', 'seq_range': [1, MAX_COUNT],
                       'elapsed_ms_range': [0, MAX_COUNT],
                       'units': 'authored integer milliseconds; no timer measurement'},
            'json_lines': lines, 'decoded_records': decoded, 'record_levels': levels,
            'created_records_for_two_valid_calls': after_valid,
            'serializer_only_probe': {'encoded': encoded_probe, 'application_event': False},
            'actual_sensor_network_disk_or_root_reconfiguration_used': False,
            'checks': checks}


if __name__ == '__main__':
    print(json.dumps(run_example(), indent=2))

۴. خروجی و هجده بررسی دقیقاً چه چیزی را می‌سنجند؟

دو خط خروجی، شماره‌های ۷ و ۸، وضعیت‌های accepted و rejected و مدت‌های ۴ و ۹ را نگه می‌دارند. هر دو پس از تجزیه دقیقاً همان چهار کلید را دارند. INFO برای اولی و WARNING برای دومی سیاست انتخابی این مثال است؛ از آن نتیجه نمی‌گیریم همهٔ ردشدن‌های ورودی در هر سامانه باید هشدار باشند. سطح در فرادادهٔ LogRecord بررسی می‌شود و جزو چهار کلید JSON نیست.

{
  "python": "3.14.5",
  "scope": "private_StringIO_synthetic_event_fixture",
  "schema": {
    "input_keys": [
      "seq",
      "status",
      "elapsed_ms"
    ],
    "output_keys": [
      "event",
      "seq",
      "status",
      "elapsed_ms"
    ],
    "event_code": "sensor.validation",
    "seq_range": [
      1,
      2147483647
    ],
    "elapsed_ms_range": [
      0,
      2147483647
    ],
    "units": "authored integer milliseconds; no timer measurement"
  },
  "json_lines": [
    "{\"event\":\"sensor.validation\",\"seq\":7,\"status\":\"accepted\",\"elapsed_ms\":4}",
    "{\"event\":\"sensor.validation\",\"seq\":8,\"status\":\"rejected\",\"elapsed_ms\":9}"
  ],
  "decoded_records": [
    {
      "event": "sensor.validation",
      "seq": 7,
      "status": "accepted",
      "elapsed_ms": 4
    },
    {
      "event": "sensor.validation",
      "seq": 8,
      "status": "rejected",
      "elapsed_ms": 9
    }
  ],
  "record_levels": [
    "INFO",
    "WARNING"
  ],
  "created_records_for_two_valid_calls": 2,
  "serializer_only_probe": {
    "encoded": "{\"probe\":\"\\u062d\\u0633\\u06af\\u0631\\n\\u0633\\u0637\\u0631 \\u062f\\u0648\\u0645\\r\\n\"}",
    "application_event": false
  },
  "actual_sensor_network_disk_or_root_reconfiguration_used": false,
  "checks": {
    "two_calls_produce_exactly_two_physical_lines": true,
    "json_roundtrip_matches_authored_records": true,
    "each_record_has_only_four_fixed_keys": true,
    "output_numeric_fields_remain_exact_integers": true,
    "status_codes_map_to_authored_info_warning_levels": true,
    "bool_sequence_rejected": true,
    "bool_elapsed_rejected": true,
    "out_of_range_sequence_and_elapsed_rejected": true,
    "float_nan_and_infinity_rejected_before_logging": true,
    "arbitrary_or_newline_status_text_rejected": true,
    "missing_unknown_and_nondict_inputs_rejected": true,
    "invalid_inputs_create_no_logrecord_or_line": true,
    "private_logger_has_one_handler_without_propagation": true,
    "input_is_not_mutated_and_serialization_is_repeatable": true,
    "zero_elapsed_and_inclusive_schema_boundaries_accepted": true,
    "separate_json_probe_escapes_controls_and_preserves_unicode": true,
    "json_nan_encoding_is_explicitly_disabled": true,
    "fresh_sink_does_not_accumulate_handlers_or_change_root_configuration": true
  }
}

هجده بررسی، رفت‌وبرگشت JSON، نوع عددها، حدود شامل انتها، رد بولی و مقدارهای نامتناهی، وضعیت نامعتبر و کلیدهای اضافی را پوشش می‌دهند. بررسی جداگانه ثابت می‌کند سریال‌ساز با allow_nan خاموش، NaN را هم رد می‌کند؛ در مسیر اصلی، نوع آن پیش‌تر رد شده است. پس این دو مشاهده را به معنی پذیرش و پاک‌سازی خودکار عدد نامعتبر معرفی نمی‌کنیم.

مقصد تازه را نیز می‌سازیم و یک بار می‌نویسیم؛ دقیقاً یک رکورد و یک خط می‌گیریم و تنظیمات خوانده‌شدهٔ root ثابت می‌مانند. این نتیجه دربارهٔ تکراری نشدن handler در همین آزمون است. اگر رویدادی را عمداً دوبار فراخوانی کنیم، دو ثبت خواهیم داشت؛ حذف تکرار تجاری پیاده نشده است. موفقیت نوشتن در StringIO هم ماندگاری دیسک، تحمل قطع شبکه یا تحویل قطعی به جمع‌آورنده را ثابت نمی‌کند.

۵. چهار تمرین پاسخ‌دار و تصمیم‌های بعد از قالب

تمرین اول: آیا True برای مدت، معادل یک میلی‌ثانیه پذیرفته می‌شود؟ پاسخ: خیر؛ قرارداد نوع دقیق int دارد و بولی با TypeError رد می‌شود. اینجا تبدیل بولی به عدد بخشی از تصمیم ما نیست.

تمرین دوم: اضافه کردن payload تو‌در‌تو چه می‌کند؟ پاسخ: مجموعهٔ کلیدها با طرح برابر نیست و پیش از ساخت رکورد خطا می‌گیریم. این مسیر، محتوای payload را داخل پیام خطا یا گزارش کپی نمی‌کند؛ ادعایی دربارهٔ مسیرهای دیگر برنامه ندارد.

تمرین سوم: آیا نمونهٔ فارسی دارای خط جدید، فیلد مجاز رویداد است؟ پاسخ: خیر؛ فقط آزمون مستقل JSON است. در وضعیت اصلی، رشتهٔ آزاد پذیرفته نمی‌شود و فراردهی جای قرارداد معنایی وضعیت را نمی‌گیرد.

تمرین چهارم: چرا دو خط درست، اثبات امنیت گزارش نیست؟ پاسخ: دسترسی، نگهداری، خرابی مقصد و رفتار سامانه‌های پایین‌دست را نسنجیده‌ایم. راهنمای طراحی گزارش رویداد سرور این تصمیم‌های گسترده‌تر را بررسی می‌کند. آزمون حاضر فقط قرارداد و خروجی مسیر حافظه‌ای مشخص را قابل بازبینی می‌سازد.

منابع و مطالعهٔ بیشتر

• Python3.14 logging؛ Logger، handler، propagate و makeRecord
• Python3.14 json؛ dumps، loads، فراردهی و allow_nan
• OWASP Logging؛ قرارداد رویداد، حذف دادهٔ غیرضروری و زمینهٔ امن شکست
• Python3.14 unittest.mock؛ patch.object و wraps برای مشاهدهٔ همان نمونه

بررسی منابع: ۲۴ مهر ۱۴۰۵. تحلیل‌ها و پیشنهادهای اجرایی، جمع‌بندی نویسنده‌اند.