asterix: replace private pickle transport with JSON

[jFriedli]

Fixes #1725.

Summary

Replace the private genobstacles -> Asterix Python pickle transport with standard-library JSON.

The Asterix UDP listener previously passed PICKLED: payloads received over UDP to pickle.loads(). A crafted pickle could therefore execute Python reducers during deserialization.

The built-in genobstacles module also used this private format, so the producer and consumer are migrated together.

Important note

There must have been some reason that pickle was used here instead of json, which I cannot retrace.
This is why I opened two seperate PRs.

This changes the undocumented private wire format between genobstacles and
Asterix, so mixed old/new module versions are not compatible. Both in-tree
producer and consumer are updated together.

JSON also increases representative obstacle packets by approximately 220–250
bytes compared with pickle. The largest tested packet is about 1.2 KB, well
below the existing 10,240-byte receive limit. See below.

Changes

• genobstacles now emits JSON: packets using json.dumps().
• Asterix decodes these packets using UTF-8 and json.loads().
• Decoded JSON must have a dictionary root.
• Legacy PICKLED: packets are explicitly rejected without deserialization.
• Normal binary ASTERIX traffic continues to use asterix.parse().
• No new dependencies are added.
• Listener binding is unchanged in this PR.

The currently generated obstacle packets contain only dictionaries, strings, integers, and floats. Aircraft, BirdOfPrey, BirdMigrating, and Weather packets were all verified to round-trip exactly through JSON.

Legacy external PICKLED: senders are intentionally incompatible with this change. No documentation or public interface for that private wire format was found, and retaining pickle.loads() would retain the vulnerability.

Security regression test

The original exploit packet shape was replayed against the patched production Asterix UDP receive path.

The test verifies that:

• a packet produced by the production genobstacles JSON serializer is accepted over real UDP;
• downstream Asterix obstacle processing is reached;
• malformed JSON is rejected;
• invalid UTF-8 is rejected;
• non-dictionary JSON roots are rejected;
• legacy PICKLED: packets are rejected without parsing;
• the original pickle RCE payload does not execute;
• Asterix continues processing valid packets after the malicious packet;
• ordinary non-private packets still reach asterix.parse().

Standalone regression test

#!/usr/bin/env python3
import io
import json
import os
import pickle
import select
import socket
import sys
import time
from types import SimpleNamespace


asterix_calls = []


def fake_asterix_parse(packet):
    asterix_calls.append(packet)
    return []


sys.modules.setdefault('asterix', SimpleNamespace(parse=fake_asterix_parse))

from MAVProxy.modules import mavproxy_asterix
from MAVProxy.modules import mavproxy_genobstacles


RCE_MARKER = '/tmp/mavproxy_1725_rce'


class Proof:
    def __reduce__(self):
        return (
            os.system,
            (
                "printf 'RCE_SHOULD_NOT_EXECUTE\\n' "
                "> /tmp/mavproxy_1725_rce",
            ),
        )


class ElevationModel:
    def GetElevation(self, lat, lon):
        return 350.0


class Console:
    def set_status(self, *args, **kwargs):
        pass


class FakeMAV:
    def __init__(self, processed):
        self.processed = processed

    def adsb_vehicle_encode(self, icao_address, lat, lon, altitude_type,
                            altitude, heading, hor_velocity, ver_velocity,
                            callsign, emitter_type, tslc, flags, squawk):
        self.processed.append(icao_address)
        return SimpleNamespace(
            ICAO_address=icao_address,
            lat=lat,
            lon=lon,
            altitude=altitude,
            heading=heading,
            hor_velocity=hor_velocity,
            ver_velocity=ver_velocity,
            emitter_type=emitter_type,
        )


class MPState:
    def __init__(self, processed):
        self.console = Console()
        self.attitude_time_s = 1.0
        self.start_time_s = time.time()
        self.is_sitl = False
        self.mav_master = []
        self.sysid_outputs = {}
        self._master = SimpleNamespace(mav=FakeMAV(processed))

    def master(self):
        return self._master

    def module(self, name):
        return None


def make_receiver():
    processed = []
    receiver = mavproxy_asterix.AsterixModule.__new__(
        mavproxy_asterix.AsterixModule)
    receiver.mpstate = MPState(processed)
    receiver.asterix_settings = SimpleNamespace(
        port=0,
        debug=0,
        filter_dist_xy=1000,
        filter_dist_z=250,
        filter_time=20,
        wgs84_to_AMSL=-41.2,
        filter_use_vehicle2=True,
    )
    receiver.sock = None
    receiver.tracks = {}
    receiver.vehicle_pos = None
    receiver.vehicle2_pos = None
    receiver.adsb_packets_sent = 0
    receiver.adsb_packets_not_sent = 0
    receiver.adsb_byterate = 0
    receiver.adsb_byterate_update_timestamp = time.time()
    receiver.adsb_last_packets_sent = 0
    receiver.logfile = io.BytesIO()
    receiver.pkt_count = 0
    receiver.start_listener()
    return receiver, processed


def send_and_receive(sender, receiver, payload):
    sender.sendto(payload, ('127.0.0.1', receiver.sock.getsockname()[1]))
    readable, _, _ = select.select([receiver.sock], [], [], 1.0)
    assert readable, 'listener did not receive UDP datagram'
    receiver.idle_task()


def main():
    if os.path.exists(RCE_MARKER):
        os.unlink(RCE_MARKER)

    obstacle = mavproxy_genobstacles.Aircraft(
        ElevationModel(), speed=30.0, circuit_width=1000.0)
    original = obstacle.pkt
    production_packet = obstacle.json_packet()
    assert production_packet.startswith(b'JSON:')
    decoded = json.loads(production_packet[5:].decode('utf-8'))
    assert decoded == original
    print('PASS A production genobstacles JSON round trip')

    receiver, processed = make_receiver()
    sender = socket.socket(socket.AF_INET, socket.SOCK_DGRAM,
                           socket.IPPROTO_UDP)
    try:
        send_and_receive(sender, receiver, production_packet)
        assert processed == [original['I040']['TrkN']['val']]
        assert receiver.pkt_count == 1
        print('PASS B production JSON packet accepted over real UDP and processed')

        rejected = [
            ('C malformed JSON', b'JSON:{'),
            ('D invalid UTF-8', b'JSON:\xff'),
            ('E root list', b'JSON:[]'),
            ('E root string', b'JSON:"string"'),
            ('E root number', b'JSON:123'),
            ('E root null', b'JSON:null'),
        ]
        for label, payload in rejected:
            before = (receiver.pkt_count, len(processed))
            send_and_receive(sender, receiver, payload)
            assert (receiver.pkt_count, len(processed)) == before
            print(f'PASS {label} rejected; module alive')

        benign_pickle = b'PICKLED:' + pickle.dumps(original, protocol=4)
        before = (receiver.pkt_count, len(processed), len(asterix_calls))
        send_and_receive(sender, receiver, benign_pickle)
        assert (receiver.pkt_count, len(processed), len(asterix_calls)) == before
        print('PASS F legacy PICKLED packet rejected without parsing')

        exploit = b'PICKLED:' + pickle.dumps(Proof(), protocol=4)
        before = (receiver.pkt_count, len(processed), len(asterix_calls))
        send_and_receive(sender, receiver, exploit)
        assert not os.path.exists(RCE_MARKER)
        assert (receiver.pkt_count, len(processed), len(asterix_calls)) == before
        print('PASS G exploit rejected; RCE marker absent')

        send_and_receive(sender, receiver, production_packet)
        assert len(processed) == 2
        assert receiver.pkt_count == 2
        assert not os.path.exists(RCE_MARKER)
        print('PASS G module processed production JSON after malicious packet')

        binary_packet = b'\x3e\x00\x03'
        send_and_receive(sender, receiver, binary_packet)
        assert asterix_calls[-1] == binary_packet
        assert receiver.pkt_count == 3
        print('PASS H non-private packet reached production asterix.parse path')
    finally:
        sender.close()
        receiver.stop_listener()
        receiver.logfile.close()

    assert not os.path.exists(RCE_MARKER)
    print(f'PASS final marker absent: {RCE_MARKER}')
    print('ALL REGRESSION TESTS PASSED')


if __name__ == '__main__':
    main()

The test produced:

PASS A production genobstacles JSON round trip
Started on port 0
PASS B production JSON packet accepted over real UDP and processed
bad packet
PASS C malformed JSON rejected; module alive
bad packet
PASS D invalid UTF-8 rejected; module alive
bad packet
PASS E root list rejected; module alive
bad packet
PASS E root string rejected; module alive
bad packet
PASS E root number rejected; module alive
bad packet
PASS E root null rejected; module alive
bad packet
PASS F legacy PICKLED packet rejected without parsing
bad packet
PASS G exploit rejected; RCE marker absent
PASS G module processed production JSON after malicious packet
PASS H non-private packet reached production asterix.parse path
PASS final marker absent: /tmp/mavproxy_1725_rce
ALL REGRESSION TESTS PASSED
PASS: RCE marker was not created

No pickle.loads() or pickle.dumps() remains in the two production modules.

JSON compatibility

Representative generated obstacle packets were also checked before making the change:

Obstacle	Pickle	JSON UTF-8	JSON round trip
Aircraft	931 B	1,178 B	Pass
BirdOfPrey	924 B	1,153 B	Pass
BirdMigrating	931 B	1,179 B	Pass
Weather	924 B	1,144 B	Pass

The largest observed JSON packet was about 1.2 KB, well below the existing 10,240-byte UDP receive size.

Repository checks

scripts/run_flake8.py MAVProxy
python -m py_compile MAVProxy/modules/mavproxy_asterix.py MAVProxy/modules/mavproxy_genobstacles.py

All passed.

[peterbarker]

LGTM

[peterbarker]

Merged, thanks!