o i(@sZdZddlZddlmZdZdZdZdZdZdZ d Z d Z d Z d Z d Zd ZGd ddZdS)a TLV (Type-Length-Value) Binary Encoder/Decoder Provides efficient binary serialization for dirty worker protocol messages. Inspired by OpenBSD msgctl/msgsnd message format. Type Codes: 0x00: None (no value bytes) 0x01: bool (1 byte: 0x00 or 0x01) 0x05: int64 (8 bytes big-endian signed) 0x06: float64 (8 bytes IEEE 754) 0x10: bytes (4-byte length + raw bytes) 0x11: string (4-byte length + UTF-8 encoded) 0x20: list (4-byte count + encoded elements) 0x21: dict (4-byte count + encoded key-value pairs) N)DirtyProtocolError !iic@sPeZdZdZedefddZed dededefdd Z edefd d Z d S) TLVEncoderz TLV binary encoder/decoder. Encodes Python values to binary TLV format and decodes back. Supports: None, bool, int, float, bytes, str, list, dict. returncCs:|dur ttgSt|trtt|rdgSdgSt|tr*ttgtd|St|t r:tt gtd|St|trat |t krRt dt |dt dttgtd t ||St|tr|d }t |tkr~t d t |dtdttgtd t ||St|ttfrt |tkrt d t |d tdttgtd t |g}|D] }|t|qd|St|trt |tkrt dt |d tdttgtd t |g}|D]\}}t|tst|}|t||t|qd|St dt|j)aJ Encode a Python value to TLV binary format. Args: value: Python value to encode (None, bool, int, float, bytes, str, list, or dict) Returns: bytes: TLV-encoded binary data Raises: DirtyProtocolError: If value type is not supported Nrr>q>dBytes too large: bytes (max: )>Iutf-8String too large: List too large: items (max: Dict too large: z#Unsupported type for TLV encoding: ) bytes TYPE_NONE isinstancebool TYPE_BOOLint TYPE_INT64structpackfloat TYPE_FLOAT64lenMAX_BYTES_SIZEr TYPE_BYTESstrencodeMAX_STRING_SIZE TYPE_STRINGlisttuple MAX_LIST_SIZE TYPE_LISTappendr joindict MAX_DICT_SIZE TYPE_DICTitemstype__name__)valueencodedpartsitemkvr<F/home/ubuntu/.local/lib/python3.10/site-packages/gunicorn/dirty/tlv.pyr'4sj                   zTLVEncoder.encoderdataoffsetc Cs|t|krtd|||dd||}|d7}|tkr"d|fS|tkrF|t|kr:td||d|dd||dk}||dfS|tkrt|dt|kr`td ||d|ddtd |||dd}||dfS|tkr|dt|krtd ||d|ddtd |||dd}||dfS|tkr |d t|krtd||d|ddtd|||d d}|d 7}|t krtd|dt d||t|krtd|dt||||d|dd||||}|||fS|t kr|d t|kr(td||d|ddtd|||d d}|d 7}|t krJtd|dt d||t|krktd|dt||||d|ddz |||| d}Wnt y}ztd||||t|ddd}~ww|||fS|tkr|d t|krtd||d|ddtd|||d d}|d 7}|tkrtd|dtdg}t|D]}t ||\} }|| q||fS|tkrg|d t|krtd||d|ddtd|||d d}|d 7}|tkr5td|dtdi} t|D]'}t ||\} }t| tsUtd t| jt ||\}}|| | <q;| |fStd!|d"||d|dd)#a9 Decode a TLV-encoded value from binary data. Args: data: Binary data to decode offset: Starting offset in the data Returns: tuple: (decoded_value, new_offset) Raises: DirtyProtocolError: If data is malformed or truncated z Truncated TLV data: no type byteraw_datarNz&Truncated TLV data: missing bool valuerz$Truncated TLV data: incomplete int64r z&Truncated TLV data: incomplete float64r z+Truncated TLV data: incomplete bytes lengthrrrrzTruncated TLV data: expected z bytes, got rz,Truncated TLV data: incomplete string lengthrz bytes for string, got rzInvalid UTF-8 in string: z)Truncated TLV data: incomplete list countrrz)Truncated TLV data: incomplete dict countrzDict key must be string, got zUnknown TLV type code: 0x02x)r#rrrrrunpackr"r%r$r)r(decodeUnicodeDecodeErrorminr-r,ranger r.r2r1rr&r4r5) r>r? type_coder6lengthecountr3_r9resultkeyr<r<r=rGs                      zTLVEncoder.decodecCsHt|d\}}|t|kr"tdt||d|||dd|S)a Decode a complete TLV-encoded value, ensuring all data is consumed. Args: data: Binary data to decode Returns: Decoded Python value Raises: DirtyProtocolError: If data is malformed or has trailing bytes rzTrailing data after TLV: z bytesr@rA)r rGr#r)r>r6r?r<r<r= decode_fulls zTLVEncoder.decode_fullN)r) r5 __module__ __qualname____doc__ staticmethodrr'rr+rGrRr<r<r<r=r ,sJr )rUrerrorsrrrrr"r%r)r-r2r(r$r,r1r r<r<r<r=s