import binascii
import logging
import os
import sys #DEBUG2**************
import ipaddress #DEBUG2**************
import time #PERF EV TIME INFO*
from coapthon.client.helperclient import HelperClient #PERF EV AUTOMATION V2*
from threading import Thread #PERF EV AUTOMATION V2*
from collections import deque
from dataclasses import dataclass
from enum import Enum
from functools import partial
from typing import Any, Deque, Dict, FrozenSet, List, Optional, Sequence, Set, Tuple
from .. import tls
from ..buffer import UINT_VAR_MAX, Buffer, BufferReadError, size_uint_var
from . import events
from .configuration import QuicConfiguration
from .crypto import CryptoError, CryptoPair, KeyUnavailableError
from .logger import QuicLoggerTrace
from .packet import (
NON_ACK_ELICITING_FRAME_TYPES,
PACKET_TYPE_HANDSHAKE,
PACKET_TYPE_INITIAL,
PACKET_TYPE_ONE_RTT,
PACKET_TYPE_RETRY,
PACKET_TYPE_ZERO_RTT,
PROBING_FRAME_TYPES,
RETRY_INTEGRITY_TAG_SIZE,
QuicErrorCode,
QuicFrameType,
QuicProtocolVersion,
QuicStreamFrame,
QuicTransportParameters,
get_retry_integrity_tag,
get_spin_bit,
is_long_header,
pull_ack_frame,
pull_quic_header,
pull_quic_transport_parameters,
push_ack_frame,
push_quic_transport_parameters,
)
from .packet_builder import (
PACKET_MAX_SIZE,
QuicDeliveryState,
QuicPacketBuilder,
QuicPacketBuilderStop,
)
from .recovery import K_GRANULARITY, QuicPacketRecovery, QuicPacketSpace
from .stream import FinalSizeError, QuicStream
logger = logging.getLogger("quic")
CRYPTO_BUFFER_SIZE = 16384
EPOCH_SHORTCUTS = {
"I": tls.Epoch.INITIAL,
"H": tls.Epoch.HANDSHAKE,
"0": tls.Epoch.ZERO_RTT,
"1": tls.Epoch.ONE_RTT,
}
MAX_EARLY_DATA = 0xFFFFFFFF
SECRETS_LABELS = [
[
None,
"QUIC_CLIENT_EARLY_TRAFFIC_SECRET",
"QUIC_CLIENT_HANDSHAKE_TRAFFIC_SECRET",
"QUIC_CLIENT_TRAFFIC_SECRET_0",
],
[
None,
None,
"QUIC_SERVER_HANDSHAKE_TRAFFIC_SECRET",
"QUIC_SERVER_TRAFFIC_SECRET_0",
],
]
STREAM_FLAGS = 0x07
NetworkAddress = Any
# frame sizes
ACK_FRAME_CAPACITY = 64 # FIXME: this is arbitrary!
APPLICATION_CLOSE_FRAME_CAPACITY = 1 + 8 + 8 # + reason length
CONNECTION_LIMIT_FRAME_CAPACITY = 1 + 8
HANDSHAKE_DONE_FRAME_CAPACITY = 1
MAX_STREAM_DATA_FRAME_CAPACITY = 1 + 8 + 8
NEW_CONNECTION_ID_FRAME_CAPACITY = 1 + 8 + 8 + 1 + 20 + 16
PATH_CHALLENGE_FRAME_CAPACITY = 1 + 8
PATH_RESPONSE_FRAME_CAPACITY = 1 + 8
PING_FRAME_CAPACITY = 1
RESET_STREAM_CAPACITY = 1 + 8 + 8 + 8
RETIRE_CONNECTION_ID_CAPACITY = 1 + 8
STREAMS_BLOCKED_CAPACITY = 1 + 8
TRANSPORT_CLOSE_FRAME_CAPACITY = 1 + 8 + 8 + 8 # + reason length
SERVER_MIGRATION_FRAME_CAPACITY = 1 + 4 #DEBUG2*
def EPOCHS(shortcut: str) -> FrozenSet[tls.Epoch]:
return frozenset(EPOCH_SHORTCUTS[i] for i in shortcut)
def dump_cid(cid: bytes) -> str:
return binascii.hexlify(cid).decode("ascii")
def get_epoch(packet_type: int) -> tls.Epoch:
if packet_type == PACKET_TYPE_INITIAL:
return tls.Epoch.INITIAL
elif packet_type == PACKET_TYPE_ZERO_RTT:
return tls.Epoch.ZERO_RTT
elif packet_type == PACKET_TYPE_HANDSHAKE:
return tls.Epoch.HANDSHAKE
else:
return tls.Epoch.ONE_RTT
def stream_is_client_initiated(stream_id: int) -> bool:
"""
Returns True if the stream is client initiated.
"""
return not (stream_id & 1)
def stream_is_unidirectional(stream_id: int) -> bool:
"""
Returns True if the stream is unidirectional.
"""
return bool(stream_id & 2)
class Limit:
def __init__(self, frame_type: int, name: str, value: int):
self.frame_type = frame_type
self.name = name
self.sent = value
self.used = 0
self.value = value
class QuicConnectionError(Exception):
def __init__(self, error_code: int, frame_type: int, reason_phrase: str):
self.error_code = error_code
self.frame_type = frame_type
self.reason_phrase = reason_phrase
def __str__(self) -> str:
s = "Error: %d, reason: %s" % (self.error_code, self.reason_phrase)
if self.frame_type is not None:
s += ", frame_type: %s" % self.frame_type
return s
class QuicConnectionAdapter(logging.LoggerAdapter):
def process(self, msg: str, kwargs: Any) -> Tuple[str, Any]:
return "[%s] %s" % (self.extra["id"], msg), kwargs
@dataclass
class QuicConnectionId:
cid: bytes
sequence_number: int
stateless_reset_token: bytes = b""
was_sent: bool = False
class QuicConnectionState(Enum):
FIRSTFLIGHT = 0
CONNECTED = 1
CLOSING = 2
DRAINING = 3
TERMINATED = 4
@dataclass
class QuicNetworkPath:
addr: NetworkAddress
bytes_received: int = 0
bytes_sent: int = 0
is_validated: bool = False
local_challenge: Optional[bytes] = None
remote_challenge: Optional[bytes] = None
def can_send(self, size: int) -> bool:
return self.is_validated or (self.bytes_sent + size) <= 3 * self.bytes_received
@dataclass
class QuicReceiveContext:
epoch: tls.Epoch
host_cid: bytes
network_path: QuicNetworkPath
quic_logger_frames: Optional[List[Any]]
time: float
END_STATES = frozenset(
[
QuicConnectionState.CLOSING,
QuicConnectionState.DRAINING,
QuicConnectionState.TERMINATED,
]
)
#PERF EV V2 AUTOMATION Client CoAP****
def clientCoAP(mtype: int, list_addr_server) -> None:
print("Before connecting to CoAP server")
client = HelperClient(server=(list_addr_server[0], 5683))
print("After connecting to CoAP server")
request = str(mtype) + ","+str(list_addr_server[0])+","+str(list_addr_server[1])
print(request)
response = client.put("basic",request)
print("Received response from CoAP server")
print(response.pretty_print())
client.stop()
#PERF EV V2 AUTOMATION Client CoAP*****
class QuicConnection:
"""
A QUIC connection.
The state machine is driven by three kinds of sources:
- the API user requesting data to be send out (see :meth:`connect`,
:meth:`reset_stream`, :meth:`send_ping`, :meth:`send_datagram_data`
and :meth:`send_stream_data`)
- data being received from the network (see :meth:`receive_datagram`)
- a timer firing (see :meth:`handle_timer`)
:param configuration: The QUIC configuration to use.
"""
def __init__(
self,
*,
configuration: QuicConfiguration,
original_destination_connection_id: Optional[bytes] = None,
retry_source_connection_id: Optional[bytes] = None,
session_ticket_fetcher: Optional[tls.SessionTicketFetcher] = None,
session_ticket_handler: Optional[tls.SessionTicketHandler] = None,
) -> None:
if configuration.is_client:
assert (
original_destination_connection_id is None
), "Cannot set original_destination_connection_id for a client"
assert (
retry_source_connection_id is None
), "Cannot set retry_source_connection_id for a client"
else:
assert (
configuration.certificate is not None
), "SSL certificate is required for a server"
assert (
configuration.private_key is not None
), "SSL private key is required for a server"
assert (
original_destination_connection_id is not None
), "original_destination_connection_id is required for a server"
# configuration
self._configuration = configuration
self._is_client = configuration.is_client
self._ack_delay = K_GRANULARITY
self._close_at: Optional[float] = None
self._close_event: Optional[events.ConnectionTerminated] = None
self._connect_called = False
self._cryptos: Dict[tls.Epoch, CryptoPair] = {}
self._crypto_buffers: Dict[tls.Epoch, Buffer] = {}
self._crypto_streams: Dict[tls.Epoch, QuicStream] = {}
self._events: Deque[events.QuicEvent] = deque()
self._handshake_complete = False
self._handshake_confirmed = False
self._host_cids = [
QuicConnectionId(
cid=os.urandom(configuration.connection_id_length),
sequence_number=0,
stateless_reset_token=os.urandom(16) if not self._is_client else None,
was_sent=True,
)
]
self.host_cid = self._host_cids[0].cid
self._host_cid_seq = 1
self._local_ack_delay_exponent = 3
self._local_active_connection_id_limit = 8
self._local_initial_source_connection_id = self._host_cids[0].cid
self._local_max_data = Limit(
frame_type=QuicFrameType.MAX_DATA,
name="max_data",
value=configuration.max_data,
)
self._local_max_stream_data_bidi_local = configuration.max_stream_data
self._local_max_stream_data_bidi_remote = configuration.max_stream_data
self._local_max_stream_data_uni = configuration.max_stream_data
self._local_max_streams_bidi = Limit(
frame_type=QuicFrameType.MAX_STREAMS_BIDI,
name="max_streams_bidi",
value=128,
)
self._local_max_streams_uni = Limit(
frame_type=QuicFrameType.MAX_STREAMS_UNI, name="max_streams_uni", value=128
)
self._loss_at: Optional[float] = None
self._network_paths: List[QuicNetworkPath] = []
self._pacing_at: Optional[float] = None
self._packet_number = 0
self._parameters_received = False
self._peer_cid = QuicConnectionId(
cid=os.urandom(configuration.connection_id_length), sequence_number=None
)
self._peer_cid_available: List[QuicConnectionId] = []
self._peer_cid_sequence_numbers: Set[int] = set([0])
self._peer_token = b""
self._quic_logger: Optional[QuicLoggerTrace] = None
self._remote_ack_delay_exponent = 3
self._remote_active_connection_id_limit = 2
self._remote_max_idle_timeout = 0.0 # seconds
self._remote_max_data = 0
self._remote_max_data_used = 0
self._remote_max_datagram_frame_size: Optional[int] = None
self._remote_max_stream_data_bidi_local = 0
self._remote_max_stream_data_bidi_remote = 0
self._remote_max_stream_data_uni = 0
self._remote_max_streams_bidi = 0
self._remote_max_streams_uni = 0
self._retry_count = 0
self._retry_source_connection_id = retry_source_connection_id
self._spaces: Dict[tls.Epoch, QuicPacketSpace] = {}
self._spin_bit = False
self._spin_highest_pn = 0
self._state = QuicConnectionState.FIRSTFLIGHT
self._streams: Dict[int, QuicStream] = {}
self._streams_blocked_bidi: List[QuicStream] = []
self._streams_blocked_uni: List[QuicStream] = []
self._version: Optional[int] = None
self._version_negotiation_count = 0
self._server_triggered_to_migrate = False #DEBUG2*
self._previous_server_address: Optional[QuicNetworkPath] = None #DEBUG2*
self._server_migration_address: Optional[QuicNetworkPath] = None #DEBUG2*
self._trigger_period = False #DEBUG2 TEST*
#self._first_time_trigger = True #DEBUG2 TEST* #DEBUG V3*
self._migration_strategy_fast = False #DEBUG V2*
self._change_addr_fast = False #DEBUG V2*
self._initial_timestamp = 0 #PERF EV TIME INFO*
self._final_timestamp = 0 #PERF EV TIME INFO*
self._migration_type = -1 #PERF EV AUTOMATION V2*
self._n_request_migration = -1 #DEBUG V3*
self._interval_migration = -1 #DEBUG V3*
self._list_addr_server = [] #DEBUG V3*
self._synched = False #Update FAST VERSION
if self._is_client:
self._original_destination_connection_id = self._peer_cid.cid
else:
self._original_destination_connection_id = (
original_destination_connection_id
)
# logging
self._logger = QuicConnectionAdapter(
logger, {"id": dump_cid(self._original_destination_connection_id)}
)
if configuration.quic_logger:
self._quic_logger = configuration.quic_logger.start_trace(
is_client=configuration.is_client,
odcid=self._original_destination_connection_id,
)
# loss recovery
self._loss = QuicPacketRecovery(
initial_rtt=configuration.initial_rtt,
peer_completed_address_validation=not self._is_client,
quic_logger=self._quic_logger,
send_probe=self._send_probe,
)
# things to send
self._close_pending = False
self._datagrams_pending: Deque[bytes] = deque()
self._handshake_done_pending = False
self._ping_pending: List[int] = []
self._probe_pending = False
self._retire_connection_ids: List[int] = []
self._streams_blocked_pending = False
# callbacks
self._session_ticket_fetcher = session_ticket_fetcher
self._session_ticket_handler = session_ticket_handler
# frame handlers
self.__frame_handlers = {
0x00: (self._handle_padding_frame, EPOCHS("IH01")),
0x01: (self._handle_ping_frame, EPOCHS("IH01")),
0x02: (self._handle_ack_frame, EPOCHS("IH1")),
0x03: (self._handle_ack_frame, EPOCHS("IH1")),
0x04: (self._handle_reset_stream_frame, EPOCHS("01")),
0x05: (self._handle_stop_sending_frame, EPOCHS("01")),
0x06: (self._handle_crypto_frame, EPOCHS("IH1")),
0x07: (self._handle_new_token_frame, EPOCHS("1")),
0x08: (self._handle_stream_frame, EPOCHS("01")),
0x09: (self._handle_stream_frame, EPOCHS("01")),
0x0A: (self._handle_stream_frame, EPOCHS("01")),
0x0B: (self._handle_stream_frame, EPOCHS("01")),
0x0C: (self._handle_stream_frame, EPOCHS("01")),
0x0D: (self._handle_stream_frame, EPOCHS("01")),
0x0E: (self._handle_stream_frame, EPOCHS("01")),
0x0F: (self._handle_stream_frame, EPOCHS("01")),
0x10: (self._handle_max_data_frame, EPOCHS("01")),
0x11: (self._handle_max_stream_data_frame, EPOCHS("01")),
0x12: (self._handle_max_streams_bidi_frame, EPOCHS("01")),
0x13: (self._handle_max_streams_uni_frame, EPOCHS("01")),
0x14: (self._handle_data_blocked_frame, EPOCHS("01")),
0x15: (self._handle_stream_data_blocked_frame, EPOCHS("01")),
0x16: (self._handle_streams_blocked_frame, EPOCHS("01")),
0x17: (self._handle_streams_blocked_frame, EPOCHS("01")),
0x18: (self._handle_new_connection_id_frame, EPOCHS("01")),
0x19: (self._handle_retire_connection_id_frame, EPOCHS("01")),
0x1A: (self._handle_path_challenge_frame, EPOCHS("01")),
0x1B: (self._handle_path_response_frame, EPOCHS("01")),
0x1C: (self._handle_connection_close_frame, EPOCHS("IH01")),
0x1D: (self._handle_connection_close_frame, EPOCHS("01")),
0x1E: (self._handle_handshake_done_frame, EPOCHS("1")),
0x30: (self._handle_datagram_frame, EPOCHS("01")),
0x31: (self._handle_datagram_frame, EPOCHS("01")),
0x32: (self._handle_server_migration_frame, EPOCHS("01")), #DEBUG2*
0x33: (self._handle_trigger_frame, EPOCHS("01")), #DEBUG2 TEST*
}
@property
def configuration(self) -> QuicConfiguration:
return self._configuration
@property
def original_destination_connection_id(self) -> bytes:
return self._original_destination_connection_id
def change_connection_id(self) -> None:
"""
Switch to the next available connection ID and retire
the previous one.
After calling this method call :meth:`datagrams_to_send` to retrieve data
which needs to be sent.
"""
if self._peer_cid_available:
# retire previous CID
self._retire_peer_cid(self._peer_cid)
# assign new CID
self._consume_peer_cid()
def close(
self,
error_code: int = QuicErrorCode.NO_ERROR,
frame_type: Optional[int] = None,
reason_phrase: str = "",
) -> None:
"""
Close the connection.
:param error_code: An error code indicating why the connection is
being closed.
:param reason_phrase: A human-readable explanation of why the
connection is being closed.
"""
if self._state not in END_STATES:
self._close_event = events.ConnectionTerminated(
error_code=error_code,
frame_type=frame_type,
reason_phrase=reason_phrase,
)
self._close_pending = True
def connect(self, addr: NetworkAddress, now: float) -> None:
"""
Initiate the TLS handshake.
This method can only be called for clients and a single time.
After calling this method call :meth:`datagrams_to_send` to retrieve data
which needs to be sent.
:param addr: The network address of the remote peer.
:param now: The current time.
"""
assert (
self._is_client and not self._connect_called
), "connect() can only be called for clients and a single time"
self._connect_called = True
self._network_paths = [QuicNetworkPath(addr, is_validated=True)]
self._version = self._configuration.supported_versions[0]
self._connect(now=now)
def datagrams_to_send(self, counter: int, hmstrategy: int, n_request_migration: int, interval_migration: int, now: float) -> List[Tuple[bytes, NetworkAddress]]: #DEBUG2 TEST DEBUG V2* PERF EV AUTOMATION* DEBUG V3*
"""
Return a list of `(data, addr)` tuples of datagrams which need to be
sent, and the network address to which they need to be sent.
After calling this method call :meth:`get_timer` to know when the next
timer needs to be set.
:param now: The current time.
"""
#PERF EV TIME INFO**** #Update FAST VERSION
#if self._is_client and not self._change_addr_fast and self._loss._pto_count == 1 and self._final_timestamp == 0: LUCA
if self._is_client and not self._migration_strategy_fast and self._loss._pto_count == 1 and self._final_timestamp == 0 and self._initial_timestamp == 0 and self._server_migration_address is not None: #Update FAST VERSION CARLO
self._initial_timestamp = time.time()
print("PRENDO TEMPO INIZIALE SLOW: " + str(self._initial_timestamp))
#PERF EV TIME INFO**** #Update FAST VERSION
network_path = self._network_paths[0]
#print(network_path) #CARLO
if self._state in END_STATES:
#print("In end state") #CARLO
return []
# build datagrams
builder = QuicPacketBuilder(
host_cid=self.host_cid,
is_client=self._is_client,
packet_number=self._packet_number,
peer_cid=self._peer_cid.cid,
peer_token=self._peer_token,
quic_logger=self._quic_logger,
spin_bit=self._spin_bit,
version=self._version,
)
if self._close_pending:
for epoch, packet_type in (
(tls.Epoch.ONE_RTT, PACKET_TYPE_ONE_RTT),
(tls.Epoch.HANDSHAKE, PACKET_TYPE_HANDSHAKE),
(tls.Epoch.INITIAL, PACKET_TYPE_INITIAL),
):
crypto = self._cryptos[epoch]
if crypto.send.is_valid():
builder.start_packet(packet_type, crypto)
self._write_connection_close_frame(
builder=builder,
epoch=epoch,
error_code=self._close_event.error_code,
frame_type=self._close_event.frame_type,
reason_phrase=self._close_event.reason_phrase,
)
self._close_pending = False
break
self._close_begin(is_initiator=True, now=now)
else:
# congestion control
builder.max_flight_bytes = (
self._loss.congestion_window - self._loss.bytes_in_flight
)
if self._probe_pending and builder.max_flight_bytes < PACKET_MAX_SIZE:
builder.max_flight_bytes = PACKET_MAX_SIZE
# limit data on un-validated network paths
if not network_path.is_validated:
#print("Network path is not validated") #CARLO
#print("Bytes received on unvalidated path until now: " +str(network_path.bytes_received)) #CARLO
#print("Bytes sent until now to unvalidated network path: " +str(network_path.bytes_sent)) #CARLO
builder.max_total_bytes = (
network_path.bytes_received * 3 - network_path.bytes_sent
)
try:
if not self._handshake_confirmed:
for epoch in [tls.Epoch.INITIAL, tls.Epoch.HANDSHAKE]:
self._write_handshake(builder, epoch, now)
self._write_application(builder, network_path, now, counter, hmstrategy, n_request_migration, interval_migration) #DEBUG2 TEST DEBUG V2* PERF EV AUTOMATION* DEBUG V3*
except QuicPacketBuilderStop:
pass
#print("Calling builder.flush") #CARLO
datagrams, packets = builder.flush()
if datagrams:
self._packet_number = builder.packet_number
# register packets
sent_handshake = False
for packet in packets:
packet.sent_time = now
self._loss.on_packet_sent(
packet=packet, space=self._spaces[packet.epoch]
)
if packet.epoch == tls.Epoch.HANDSHAKE:
sent_handshake = True
# log packet
if self._quic_logger is not None:
self._quic_logger.log_event(
category="transport",
event="packet_sent",
data={
"packet_type": self._quic_logger.packet_type(
packet.packet_type
),
"header": {
"packet_number": str(packet.packet_number),
"packet_size": packet.sent_bytes,
"scid": dump_cid(self.host_cid)
if is_long_header(packet.packet_type)
else "",
"dcid": dump_cid(self._peer_cid.cid),
},
"frames": packet.quic_logger_frames,
},
)
# check if we can discard initial keys
if sent_handshake and self._is_client:
self._discard_epoch(tls.Epoch.INITIAL)
#now = time.time() #CARLO
#print("PREPARE DATAGRAM TO SEND AT: " + str(now)) #DEBUG* CARLO
#else:
#now = time.time()
#print("No datagrams to send at: " +str(now)) #CARLO
# return datagrams to send and the destination network address
ret = []
for datagram in datagrams:
byte_length = len(datagram)
network_path.bytes_sent += byte_length
'''
#DEBUG2*************
if self._is_client and self._loss._pto_count > 2 and self._server_migration_address is not None:
ret.append((datagram,self._server_migration_address.addr))
else:
ret.append((datagram, network_path.addr))
#DEBUG2*************
'''
#DEBUG2 DEBUG V2 - Update FAST VERSION CARLO*************
if self._is_client and not self._migration_strategy_fast and self._loss._pto_count > 0 and self._server_migration_address is not None:
ret.append((datagram,self._server_migration_address.addr)) #case of Explicit - Loss during migration (after a packet to the old is lost)
print("1 " +str(self._server_migration_address.addr))
elif self._is_client and self._synched and self._server_migration_address is not None:
ret.append((datagram,self._server_migration_address.addr)) #case of Explicit - No Loss during migration (after ack to the SERVER MIGRATION FRAME was sent)
print("2 " +str(self._server_migration_address.addr))
else:
ret.append((datagram, network_path.addr)) #normal situation of both client and server
print("3 " +str(network_path.addr))
#DEBUG2 DEBUG V2 - Update FAST VERSION CARLO*************
if self._quic_logger is not None:
self._quic_logger.log_event(
category="transport",
event="datagrams_sent",
data={"byte_length": byte_length, "count": 1},
)
if self._change_addr_fast:
#Update FAST VERSION CARLO
#this is used in the NoLoss version to indicate that ack to the SERVER MIGRATION frame was sent to the old address of the server
#from now on and until reception of non-probing frame from new server address, packets will be sent to the server_migration_address
self._synched = True
self._change_addr_fast = False
# Update FAST VERSION CARLO - let the ack to the server migration frame be sent before changing address
if self._is_client and self._migration_strategy_fast and not self._synched and self._server_migration_address is not None:
#print("SERVER MIGRATION FRAME received - ready to send ack")
self._change_addr_fast = True
return ret
def get_next_available_stream_id(self, is_unidirectional=False) -> int:
"""
Return the stream ID for the next stream created by this endpoint.
"""
stream_id = (int(is_unidirectional) << 1) | int(not self._is_client)
while stream_id in self._streams:
stream_id += 4
return stream_id
def get_timer(self) -> Optional[float]:
"""
Return the time at which the timer should fire or None if no timer is needed.
"""
timer_at = self._close_at
if self._state not in END_STATES:
# ack timer
for space in self._loss.spaces:
if space.ack_at is not None and space.ack_at < timer_at and self._previous_server_address is None: #self._network_paths[0].is_validated: #DEBUG2*
timer_at = space.ack_at
# loss detection timer
self._loss_at = self._loss.get_loss_detection_time()
if self._loss_at is not None and self._loss_at < timer_at:
timer_at = self._loss_at
# pacing timer
if self._pacing_at is not None and self._pacing_at < timer_at:
timer_at = self._pacing_at
return timer_at
def handle_timer(self, now: float) -> None:
"""
Handle the timer.
After calling this method call :meth:`datagrams_to_send` to retrieve data
which needs to be sent.
:param now: The current time.
"""
# end of closing period or idle timeout
if now >= self._close_at:
if self._close_event is None:
self._close_event = events.ConnectionTerminated(
error_code=QuicErrorCode.INTERNAL_ERROR,
frame_type=None,
reason_phrase="Idle timeout",
)
self._close_end()
return
# loss detection timeout
if self._loss_at is not None and now >= self._loss_at:
self._logger.debug("Loss detection triggered")
self._loss.on_loss_detection_timeout(now=now)
def next_event(self) -> Optional[events.QuicEvent]:
"""
Retrieve the next event from the event buffer.
Returns `None` if there are no buffered events.
"""
try:
return self._events.popleft()
except IndexError:
return None
def receive_datagram(self, data: bytes, addr: NetworkAddress, now: float) -> None: #PERF EV AUTOMATION V2*
"""
Handle an incoming datagram.
After calling this method call :meth:`datagrams_to_send` to retrieve data
which needs to be sent.
:param data: The datagram which was received.
:param addr: The network address from which the datagram was received.
:param now: The current time.
"""
# stop handling packets when closing
if self._state in END_STATES:
return
print("RECEIVED DATAGRAM")
'''
print("PROCESS REGULAR PACKET FROM ") #DEBUG*
print(addr)
print("DESTINATION CID ") #DEBUG*
print(self._peer_cid.cid)
print("SOURCE CID ")
print(self.host_cid)
print("SEQUENCE NUM ")
print(self._host_cid_seq) #DEBUG*
for conn_id in self._host_cids: #DEBUG*
print("LIST SOURCE CID ")
print(conn_id) #DEBUG*
for elem_path in self._network_paths: #DEBUG*
print("LIST OF PATH")
print(elem_path) #DEBUG*
'''
#PERF EV TIME INFO****
if not self._is_client and self._initial_timestamp == 0:
self._initial_timestamp = time.time()
#PERF EV TIME INFO****
#PERF EV AUTOMATION V2***** DEBUG V3********
if not self._is_client and self._migration_type == -1 and len(self._list_addr_server) == 0:
print("READING SERVER INFO")
f = open("/home/osboxes/aioquic-explicit_UniPisa/MigrationInformation.txt", "r+") #/home/Trigger_v4/
lines = f.readlines()
c_line = 0
for linefull in lines:
line = linefull.strip()
if c_line == 0:
self._migration_type = int(line)
else:
self._list_addr_server.append(line)
c_line=c_line+1
f.truncate(0)
#print("MIGRATION INFORMATION")
#print("MIGRATION TYPE" + str(self._migration_type))
#print("LIST ADDRESSES SERVER" + self._list_addr_server)
#PERF EV AUTOMATION V2***** DEBUG V3********
if self._quic_logger is not None:
self._quic_logger.log_event(
category="transport",
event="datagrams_received",
data={"byte_length": len(data), "count": 1},
)
buf = Buffer(data=data)
while not buf.eof():
start_off = buf.tell()
try:
header = pull_quic_header(
buf, host_cid_length=self._configuration.connection_id_length
)
except ValueError:
return
# check destination CID matches
destination_cid_seq: Optional[int] = None
for connection_id in self._host_cids:
if header.destination_cid == connection_id.cid:
destination_cid_seq = connection_id.sequence_number
break
if self._is_client and destination_cid_seq is None:
if self._quic_logger is not None:
self._quic_logger.log_event(
category="transport",
event="packet_dropped",
data={"trigger": "unknown_connection_id"},
)
return
# check protocol version
if (
self._is_client
and self._state == QuicConnectionState.FIRSTFLIGHT
and header.version == QuicProtocolVersion.NEGOTIATION
and not self._version_negotiation_count
):
# version negotiation
versions = []
while not buf.eof():
versions.append(buf.pull_uint32())
if self._quic_logger is not None:
self._quic_logger.log_event(
category="transport",
event="packet_received",
data={
"packet_type": "version_negotiation",
"header": {
"scid": dump_cid(header.source_cid),
"dcid": dump_cid(header.destination_cid),
},
"frames": [],
},
)
if self._version in versions:
self._logger.warning(
"Version negotiation packet contains %s" % self._version
)
return
common = set(self._configuration.supported_versions).intersection(
versions
)
if not common:
self._logger.error("Could not find a common protocol version")
self._close_event = events.ConnectionTerminated(
error_code=QuicErrorCode.INTERNAL_ERROR,
frame_type=None,
reason_phrase="Could not find a common protocol version",
)
self._close_end()
return
self._version = QuicProtocolVersion(max(common))
self._version_negotiation_count += 1
self._logger.info("Retrying with %s", self._version)
self._connect(now=now)
return
elif (
header.version is not None
and header.version not in self._configuration.supported_versions
):
# unsupported version
if self._quic_logger is not None:
self._quic_logger.log_event(
category="transport",
event="packet_dropped",
data={"trigger": "unsupported_version"},
)
return
if self._is_client and header.packet_type == PACKET_TYPE_RETRY:
# calculate retry integrity tag
integrity_tag = get_retry_integrity_tag(
buf.data_slice(start_off, buf.tell() - RETRY_INTEGRITY_TAG_SIZE),
self._peer_cid.cid,
version=header.version,
)
if (
header.destination_cid == self.host_cid
and header.integrity_tag == integrity_tag
and not self._retry_count
):
if self._quic_logger is not None:
self._quic_logger.log_event(
category="transport",
event="packet_received",
data={
"packet_type": "retry",
"header": {
"scid": dump_cid(header.source_cid),
"dcid": dump_cid(header.destination_cid),
},
"frames": [],
},
)
self._peer_cid.cid = header.source_cid
self._peer_token = header.token
self._retry_count += 1
self._retry_source_connection_id = header.source_cid
self._logger.info(
"Retrying with token (%d bytes)" % len(header.token)
)
self._connect(now=now)
return
network_path = self._find_network_path(addr)
# server initialization
if not self._is_client and self._state == QuicConnectionState.FIRSTFLIGHT:
assert (
header.packet_type == PACKET_TYPE_INITIAL
), "first packet must be INITIAL"
self._network_paths = [network_path]
self._version = QuicProtocolVersion(header.version)
self._initialize(header.destination_cid)
# determine crypto and packet space
epoch = get_epoch(header.packet_type)
crypto = self._cryptos[epoch]
if epoch == tls.Epoch.ZERO_RTT:
space = self._spaces[tls.Epoch.ONE_RTT]
else:
space = self._spaces[epoch]
# decrypt packet
encrypted_off = buf.tell() - start_off
end_off = buf.tell() + header.rest_length
buf.seek(end_off)
try:
plain_header, plain_payload, packet_number = crypto.decrypt_packet(
data[start_off:end_off], encrypted_off, space.expected_packet_number
)
except KeyUnavailableError as exc:
self._logger.debug(exc)
if self._quic_logger is not None:
self._quic_logger.log_event(
category="transport",
event="packet_dropped",
data={"trigger": "key_unavailable"},
)
continue
except CryptoError as exc:
self._logger.debug(exc)
if self._quic_logger is not None:
self._quic_logger.log_event(
category="transport",
event="packet_dropped",
data={"trigger": "payload_decrypt_error"},
)
continue
# check reserved bits
if header.is_long_header:
reserved_mask = 0x0C
else:
reserved_mask = 0x18
if plain_header[0] & reserved_mask:
self.close(
error_code=QuicErrorCode.PROTOCOL_VIOLATION,
frame_type=None,
reason_phrase="Reserved bits must be zero",
)
return
# raise expected packet number
if packet_number > space.expected_packet_number:
space.expected_packet_number = packet_number + 1
# log packet
quic_logger_frames: Optional[List[Dict]] = None
if self._quic_logger is not None:
quic_logger_frames = []
self._quic_logger.log_event(
category="transport",
event="packet_received",
data={
"packet_type": self._quic_logger.packet_type(
header.packet_type
),
"header": {
"packet_number": str(packet_number),
"packet_size": end_off - start_off,
"dcid": dump_cid(header.destination_cid),
"scid": dump_cid(header.source_cid),
},
"frames": quic_logger_frames,
},
)
# discard initial keys and packet space
if not self._is_client and epoch == tls.Epoch.HANDSHAKE:
self._discard_epoch(tls.Epoch.INITIAL)
# update state
if self._peer_cid.sequence_number is None:
self._peer_cid.cid = header.source_cid
self._peer_cid.sequence_number = 0
if self._state == QuicConnectionState.FIRSTFLIGHT:
self._set_state(QuicConnectionState.CONNECTED)
# update spin bit
if not header.is_long_header and packet_number > self._spin_highest_pn:
spin_bit = get_spin_bit(plain_header[0])
if self._is_client:
self._spin_bit = not spin_bit
else:
self._spin_bit = spin_bit
self._spin_highest_pn = packet_number
if self._quic_logger is not None:
self._quic_logger.log_event(
category="connectivity",
event="spin_bit_updated",
data={"state": self._spin_bit},
)
# handle payload
context = QuicReceiveContext(
epoch=epoch,
host_cid=header.destination_cid,
network_path=network_path,
quic_logger_frames=quic_logger_frames,
time=now,
)
try:
is_ack_eliciting, is_probing = self._payload_received(
context, plain_payload
)
except QuicConnectionError as exc:
self._logger.warning(exc)
self.close(
error_code=exc.error_code,
frame_type=exc.frame_type,
reason_phrase=exc.reason_phrase,
)
if self._state in END_STATES or self._close_pending:
return
# update idle timeout
self._close_at = now + self._configuration.idle_timeout
# handle migration
if (
not self._is_client
and context.host_cid != self.host_cid
and epoch == tls.Epoch.ONE_RTT
):
self._logger.debug(
"Peer switching to CID %s (%d)",
dump_cid(context.host_cid),
destination_cid_seq,
)
self.host_cid = context.host_cid
self.change_connection_id()
# update network path
if not network_path.is_validated and epoch == tls.Epoch.HANDSHAKE:
self._logger.debug(
"Network path %s validated by handshake", network_path.addr
)
network_path.is_validated = True
network_path.bytes_received += end_off - start_off
if network_path not in self._network_paths:
self._network_paths.append(network_path)
idx = self._network_paths.index(network_path)
#DEBUG2 TEST* Update FAST VERSION CARLO - client case
if self._is_client and idx and not is_probing and packet_number > space.largest_received_packet and self._server_migration_address is not None:
if network_path.addr == self._server_migration_address.addr: #good situation - accept datagram and set new address as primary
self._network_paths.pop(idx)
self._network_paths.insert(0, network_path) #set to primary
print("New primary address was set")
print(str(self._network_paths[0]))
#self._network_paths.remove(self._previous_server_address)
if self._network_paths[0].is_validated:
if self._previous_server_address is not None: #DEBUG2*
print("MIGRATION SUCCESSFULLY TERMINATED") #DEBUG2*
self._previous_server_address = None #DEBUG2*
self._server_migration_address = None #DEBUG2*
if self._migration_strategy_fast: #case Explicit - No Loss
self._synched = False
if not self._migration_strategy_fast: # case R-Explicit
#PERF EV TIME INFO****
if self._initial_timestamp != 0:
self._final_timestamp = time.time()
print("DELTA SERVER MIGRATION:")
print(self._final_timestamp - self._initial_timestamp)
self._initial_timestamp = 0
self._final_timestamp = 0
#PERF EV TIME INFO****
#DEBUG2 TEST* Update FAST VERSION CARLO - server case
elif not self._is_client and idx and not is_probing and packet_number > space.largest_received_packet:
self._network_paths.pop(idx)
self._network_paths.insert(0, network_path) #set to primary
# record packet as received
if not space.discarded:
if packet_number > space.largest_received_packet:
space.largest_received_packet = packet_number
space.largest_received_time = now
space.ack_queue.add(packet_number)
if is_ack_eliciting and space.ack_at is None:
space.ack_at = now + self._ack_delay
def request_key_update(self) -> None:
"""
Request an update of the encryption keys.
"""
assert self._handshake_complete, "cannot change key before handshake completes"
self._cryptos[tls.Epoch.ONE_RTT].update_key()
def reset_stream(self, stream_id: int, error_code: int) -> None:
"""
Abruptly terminate the sending part of a stream.
:param stream_id: The stream's ID.
:param error_code: An error code indicating why the stream is being reset.
"""
stream = self._get_or_create_stream_for_send(stream_id)
stream.reset(error_code)
def send_ping(self, uid: int) -> None:
"""
Send a PING frame to the peer.
:param uid: A unique ID for this PING.
"""
self._ping_pending.append(uid)
def send_datagram_frame(self, data: bytes) -> None:
"""
Send a DATAGRAM frame.
:param data: The data to be sent.
"""
self._datagrams_pending.append(data)
def send_stream_data(
self, stream_id: int, data: bytes, end_stream: bool = False
) -> None:
"""
Send data on the specific stream.
:param stream_id: The stream's ID.
:param data: The data to be sent.
:param end_stream: If set to `True`, the FIN bit will be set.
"""
stream = self._get_or_create_stream_for_send(stream_id)
stream.write(data, end_stream=end_stream)
# Private
def _alpn_handler(self, alpn_protocol: str) -> None:
"""
Callback which is invoked by the TLS engine when ALPN negotiation completes.
"""
self._events.append(events.ProtocolNegotiated(alpn_protocol=alpn_protocol))
def _assert_stream_can_receive(self, frame_type: int, stream_id: int) -> None:
"""
Check the specified stream can receive data or raises a QuicConnectionError.
"""
if not self._stream_can_receive(stream_id):
raise QuicConnectionError(
error_code=QuicErrorCode.STREAM_STATE_ERROR,
frame_type=frame_type,
reason_phrase="Stream is send-only",
)
def _assert_stream_can_send(self, frame_type: int, stream_id: int) -> None:
"""
Check the specified stream can send data or raises a QuicConnectionError.
"""
if not self._stream_can_send(stream_id):
raise QuicConnectionError(
error_code=QuicErrorCode.STREAM_STATE_ERROR,
frame_type=frame_type,
reason_phrase="Stream is receive-only",
)
def _consume_peer_cid(self) -> None:
"""
Update the destination connection ID by taking the next
available connection ID provided by the peer.
"""
self._peer_cid = self._peer_cid_available.pop(0)
self._logger.debug(
"Switching to CID %s (%d)",
dump_cid(self._peer_cid.cid),
self._peer_cid.sequence_number,
)
def _close_begin(self, is_initiator: bool, now: float) -> None:
"""
Begin the close procedure.
"""
self._close_at = now + 3 * self._loss.get_probe_timeout()
if is_initiator:
self._set_state(QuicConnectionState.CLOSING)
else:
self._set_state(QuicConnectionState.DRAINING)
def _close_end(self) -> None:
"""
End the close procedure.
"""
self._close_at = None
for epoch in self._spaces.keys():
self._discard_epoch(epoch)
self._events.append(self._close_event)
self._set_state(QuicConnectionState.TERMINATED)
# signal log end
if self._quic_logger is not None:
self._configuration.quic_logger.end_trace(self._quic_logger)
self._quic_logger = None
def _connect(self, now: float) -> None:
"""
Start the client handshake.
"""
assert self._is_client
self._close_at = now + self._configuration.idle_timeout
self._initialize(self._peer_cid.cid)
self.tls.handle_message(b"", self._crypto_buffers)
self._push_crypto_data()
def _discard_epoch(self, epoch: tls.Epoch) -> None:
if not self._spaces[epoch].discarded:
self._logger.debug("Discarding epoch %s", epoch)
self._cryptos[epoch].teardown()
self._loss.discard_space(self._spaces[epoch])
self._spaces[epoch].discarded = True
def _find_network_path(self, addr: NetworkAddress) -> QuicNetworkPath:
# check existing network paths
for idx, network_path in enumerate(self._network_paths):
if network_path.addr == addr:
return network_path
# new network path
network_path = QuicNetworkPath(addr)
self._logger.debug("Network path %s discovered", network_path.addr)
return network_path
def _get_or_create_stream(self, frame_type: int, stream_id: int) -> QuicStream:
"""
Get or create a stream in response to a received frame.
"""
stream = self._streams.get(stream_id, None)
if stream is None:
# check initiator
if stream_is_client_initiated(stream_id) == self._is_client:
raise QuicConnectionError(
error_code=QuicErrorCode.STREAM_STATE_ERROR,
frame_type=frame_type,
reason_phrase="Wrong stream initiator",
)
# determine limits
if stream_is_unidirectional(stream_id):
max_stream_data_local = self._local_max_stream_data_uni
max_stream_data_remote = 0
max_streams = self._local_max_streams_uni
else:
max_stream_data_local = self._local_max_stream_data_bidi_remote
max_stream_data_remote = self._remote_max_stream_data_bidi_local
max_streams = self._local_max_streams_bidi
# check max streams
stream_count = (stream_id // 4) + 1
if stream_count > max_streams.value:
raise QuicConnectionError(
error_code=QuicErrorCode.STREAM_LIMIT_ERROR,
frame_type=frame_type,
reason_phrase="Too many streams open",
)
elif stream_count > max_streams.used:
max_streams.used = stream_count
# create stream
self._logger.debug("Stream %d created by peer" % stream_id)
stream = self._streams[stream_id] = QuicStream(
stream_id=stream_id,
max_stream_data_local=max_stream_data_local,
max_stream_data_remote=max_stream_data_remote,
)
return stream
def _get_or_create_stream_for_send(self, stream_id: int) -> QuicStream:
"""
Get or create a QUIC stream in order to send data to the peer.
This always occurs as a result of an API call.
"""
if stream_is_client_initiated(stream_id) != self._is_client:
if stream_id not in self._streams:
raise ValueError("Cannot send data on unknown peer-initiated stream")
if stream_is_unidirectional(stream_id):
raise ValueError(
"Cannot send data on peer-initiated unidirectional stream"
)
stream = self._streams.get(stream_id, None)
if stream is None:
# determine limits
if stream_is_unidirectional(stream_id):
max_stream_data_local = 0
max_stream_data_remote = self._remote_max_stream_data_uni
max_streams = self._remote_max_streams_uni
streams_blocked = self._streams_blocked_uni
else:
max_stream_data_local = self._local_max_stream_data_bidi_local
max_stream_data_remote = self._remote_max_stream_data_bidi_remote
max_streams = self._remote_max_streams_bidi
streams_blocked = self._streams_blocked_bidi
# create stream
stream = self._streams[stream_id] = QuicStream(
stream_id=stream_id,
max_stream_data_local=max_stream_data_local,
max_stream_data_remote=max_stream_data_remote,
)
# mark stream as blocked if needed
if stream_id // 4 >= max_streams:
stream.is_blocked = True
streams_blocked.append(stream)
self._streams_blocked_pending = True
return stream
def _handle_session_ticket(self, session_ticket: tls.SessionTicket) -> None:
if (
session_ticket.max_early_data_size is not None
and session_ticket.max_early_data_size != MAX_EARLY_DATA
):
raise QuicConnectionError(
error_code=QuicErrorCode.PROTOCOL_VIOLATION,
frame_type=QuicFrameType.CRYPTO,
reason_phrase="Invalid max_early_data value %s"
% session_ticket.max_early_data_size,
)
self._session_ticket_handler(session_ticket)
def _initialize(self, peer_cid: bytes) -> None:
# TLS
self.tls = tls.Context(
alpn_protocols=self._configuration.alpn_protocols,
cadata=self._configuration.cadata,
cafile=self._configuration.cafile,
capath=self._configuration.capath,
cipher_suites=self.configuration.cipher_suites,
is_client=self._is_client,
logger=self._logger,
max_early_data=None if self._is_client else MAX_EARLY_DATA,
server_name=self._configuration.server_name,
verify_mode=self._configuration.verify_mode,
)
self.tls.certificate = self._configuration.certificate
self.tls.certificate_chain = self._configuration.certificate_chain
self.tls.certificate_private_key = self._configuration.private_key
self.tls.handshake_extensions = [
(
tls.ExtensionType.QUIC_TRANSPORT_PARAMETERS,
self._serialize_transport_parameters(),
)
]
# TLS session resumption
session_ticket = self._configuration.session_ticket
if (
self._is_client
and session_ticket is not None
and session_ticket.is_valid
and session_ticket.server_name == self._configuration.server_name
):
self.tls.session_ticket = self._configuration.session_ticket
# parse saved QUIC transport parameters - for 0-RTT
if session_ticket.max_early_data_size == MAX_EARLY_DATA:
for ext_type, ext_data in session_ticket.other_extensions:
if ext_type == tls.ExtensionType.QUIC_TRANSPORT_PARAMETERS:
self._parse_transport_parameters(
ext_data, from_session_ticket=True
)
break
# TLS callbacks
self.tls.alpn_cb = self._alpn_handler
if self._session_ticket_fetcher is not None:
self.tls.get_session_ticket_cb = self._session_ticket_fetcher
if self._session_ticket_handler is not None:
self.tls.new_session_ticket_cb = self._handle_session_ticket
self.tls.update_traffic_key_cb = self._update_traffic_key
# packet spaces
def create_crypto_pair(epoch: tls.Epoch) -> CryptoPair:
epoch_name = ["initial", "0rtt", "handshake", "1rtt"][epoch.value]
secret_names = [
"server_%s_secret" % epoch_name,
"client_%s_secret" % epoch_name,
]
recv_secret_name = secret_names[not self._is_client]
send_secret_name = secret_names[self._is_client]
return CryptoPair(
recv_setup_cb=partial(self._log_key_updated, recv_secret_name),
recv_teardown_cb=partial(self._log_key_retired, recv_secret_name),
send_setup_cb=partial(self._log_key_updated, send_secret_name),
send_teardown_cb=partial(self._log_key_retired, send_secret_name),
)
self._cryptos = dict(
(epoch, create_crypto_pair(epoch))
for epoch in (
tls.Epoch.INITIAL,
tls.Epoch.ZERO_RTT,
tls.Epoch.HANDSHAKE,
tls.Epoch.ONE_RTT,
)
)
self._crypto_buffers = {
tls.Epoch.INITIAL: Buffer(capacity=CRYPTO_BUFFER_SIZE),
tls.Epoch.HANDSHAKE: Buffer(capacity=CRYPTO_BUFFER_SIZE),
tls.Epoch.ONE_RTT: Buffer(capacity=CRYPTO_BUFFER_SIZE),
}
self._crypto_streams = {
tls.Epoch.INITIAL: QuicStream(),
tls.Epoch.HANDSHAKE: QuicStream(),
tls.Epoch.ONE_RTT: QuicStream(),
}
self._spaces = {
tls.Epoch.INITIAL: QuicPacketSpace(),
tls.Epoch.HANDSHAKE: QuicPacketSpace(),
tls.Epoch.ONE_RTT: QuicPacketSpace(),
}
self._cryptos[tls.Epoch.INITIAL].setup_initial(
cid=peer_cid, is_client=self._is_client, version=self._version
)
self._loss.spaces = list(self._spaces.values())
self._packet_number = 0
def _handle_ack_frame(
self, context: QuicReceiveContext, frame_type: int, buf: Buffer
) -> None:
"""
Handle an ACK frame.
"""
ack_rangeset, ack_delay_encoded = pull_ack_frame(buf)
if frame_type == QuicFrameType.ACK_ECN:
buf.pull_uint_var()
buf.pull_uint_var()
buf.pull_uint_var()
ack_delay = (ack_delay_encoded << self._remote_ack_delay_exponent) / 1000000
# log frame
if self._quic_logger is not None:
context.quic_logger_frames.append(
self._quic_logger.encode_ack_frame(ack_rangeset, ack_delay)
)
# check whether peer completed address validation
if not self._loss.peer_completed_address_validation and context.epoch in (
tls.Epoch.HANDSHAKE,
tls.Epoch.ONE_RTT,
):
self._loss.peer_completed_address_validation = True
self._loss.on_ack_received(
space=self._spaces[context.epoch],
ack_rangeset=ack_rangeset,
ack_delay=ack_delay,
now=context.time,
)
def _handle_connection_close_frame(
self, context: QuicReceiveContext, frame_type: int, buf: Buffer
) -> None:
"""
Handle a CONNECTION_CLOSE frame.
"""
error_code = buf.pull_uint_var()
if frame_type == QuicFrameType.TRANSPORT_CLOSE:
frame_type = buf.pull_uint_var()
else:
frame_type = None
reason_length = buf.pull_uint_var()
try:
reason_phrase = buf.pull_bytes(reason_length).decode("utf8")
except UnicodeDecodeError:
reason_phrase = ""
# log frame
if self._quic_logger is not None:
context.quic_logger_frames.append(
self._quic_logger.encode_connection_close_frame(
error_code=error_code,
frame_type=frame_type,
reason_phrase=reason_phrase,
)
)
#PERF EV TIME INFO****
if not self._is_client and self._initial_timestamp != 0:
self._final_timestamp = time.time()
print("TIME OF THE ALL CONNECTION")
print(self._final_timestamp - self._initial_timestamp)
#PERF EV TIME INFO****
self._logger.info(
"Connection close code 0x%X, reason %s", error_code, reason_phrase
)
self._close_event = events.ConnectionTerminated(
error_code=error_code, frame_type=frame_type, reason_phrase=reason_phrase
)
self._close_begin(is_initiator=False, now=context.time)
def _handle_crypto_frame(
self, context: QuicReceiveContext, frame_type: int, buf: Buffer
) -> None:
"""
Handle a CRYPTO frame.
"""
offset = buf.pull_uint_var()
length = buf.pull_uint_var()
if offset + length > UINT_VAR_MAX:
raise QuicConnectionError(
error_code=QuicErrorCode.FRAME_ENCODING_ERROR,
frame_type=frame_type,
reason_phrase="offset + length cannot exceed 2^62 - 1",
)
frame = QuicStreamFrame(offset=offset, data=buf.pull_bytes(length))
# log frame
if self._quic_logger is not None:
context.quic_logger_frames.append(
self._quic_logger.encode_crypto_frame(frame)
)
stream = self._crypto_streams[context.epoch]
event = stream.add_frame(frame)
if event is not None:
# pass data to TLS layer
try:
self.tls.handle_message(event.data, self._crypto_buffers)
self._push_crypto_data()
except tls.Alert as exc:
raise QuicConnectionError(
error_code=QuicErrorCode.CRYPTO_ERROR + int(exc.description),
frame_type=frame_type,
reason_phrase=str(exc),
)
# parse transport parameters
if (
not self._parameters_received
and self.tls.received_extensions is not None
):
for ext_type, ext_data in self.tls.received_extensions:
if ext_type == tls.ExtensionType.QUIC_TRANSPORT_PARAMETERS:
self._parse_transport_parameters(ext_data)
self._parameters_received = True
break
assert (
self._parameters_received
), "No QUIC transport parameters received"
# update current epoch
if not self._handshake_complete and self.tls.state in [
tls.State.CLIENT_POST_HANDSHAKE,
tls.State.SERVER_POST_HANDSHAKE,
]:
self._handshake_complete = True
# for servers, the handshake is now confirmed
if not self._is_client:
self._discard_epoch(tls.Epoch.HANDSHAKE)
self._handshake_confirmed = True
self._handshake_done_pending = True
self._replenish_connection_ids()
self._events.append(
events.HandshakeCompleted(
alpn_protocol=self.tls.alpn_negotiated,
early_data_accepted=self.tls.early_data_accepted,
session_resumed=self.tls.session_resumed,
)
)
self._unblock_streams(is_unidirectional=False)
self._unblock_streams(is_unidirectional=True)
self._logger.info(
"ALPN negotiated protocol %s", self.tls.alpn_negotiated
)
def _handle_data_blocked_frame(
self, context: QuicReceiveContext, frame_type: int, buf: Buffer
) -> None:
"""
Handle a DATA_BLOCKED frame.
"""
limit = buf.pull_uint_var()
# log frame
if self._quic_logger is not None:
context.quic_logger_frames.append(
self._quic_logger.encode_data_blocked_frame(limit=limit)
)
def _handle_datagram_frame(
self, context: QuicReceiveContext, frame_type: int, buf: Buffer
) -> None:
"""
Handle a DATAGRAM frame.
"""
start = buf.tell()
if frame_type == QuicFrameType.DATAGRAM_WITH_LENGTH:
length = buf.pull_uint_var()
else:
length = buf.capacity - start
data = buf.pull_bytes(length)
# log frame
if self._quic_logger is not None:
context.quic_logger_frames.append(
self._quic_logger.encode_datagram_frame(length=length)
)
# check frame is allowed
if (
self._configuration.max_datagram_frame_size is None
or buf.tell() - start >= self._configuration.max_datagram_frame_size
):
raise QuicConnectionError(
error_code=QuicErrorCode.PROTOCOL_VIOLATION,
frame_type=frame_type,
reason_phrase="Unexpected DATAGRAM frame",
)
self._events.append(events.DatagramFrameReceived(data=data))
def _handle_handshake_done_frame(
self, context: QuicReceiveContext, frame_type: int, buf: Buffer
) -> None:
"""
Handle a HANDSHAKE_DONE frame.
"""
# log frame
if self._quic_logger is not None:
context.quic_logger_frames.append(
self._quic_logger.encode_handshake_done_frame()
)
if not self._is_client:
raise QuicConnectionError(
error_code=QuicErrorCode.PROTOCOL_VIOLATION,
frame_type=frame_type,
reason_phrase="Clients must not send HANDSHAKE_DONE frames",
)
# for clients, the handshake is now confirmed
if not self._handshake_confirmed:
self._discard_epoch(tls.Epoch.HANDSHAKE)
self._handshake_confirmed = True
self._loss.peer_completed_address_validation = True
def _handle_max_data_frame(
self, context: QuicReceiveContext, frame_type: int, buf: Buffer
) -> None:
"""
Handle a MAX_DATA frame.
This adjusts the total amount of we can send to the peer.
"""
max_data = buf.pull_uint_var()
# log frame
if self._quic_logger is not None:
context.quic_logger_frames.append(
self._quic_logger.encode_connection_limit_frame(
frame_type=frame_type, maximum=max_data
)
)
if max_data > self._remote_max_data:
self._logger.debug("Remote max_data raised to %d", max_data)
self._remote_max_data = max_data
def _handle_max_stream_data_frame(
self, context: QuicReceiveContext, frame_type: int, buf: Buffer
) -> None:
"""
Handle a MAX_STREAM_DATA frame.
This adjusts the amount of data we can send on a specific stream.
"""
stream_id = buf.pull_uint_var()
max_stream_data = buf.pull_uint_var()
# log frame
if self._quic_logger is not None:
context.quic_logger_frames.append(
self._quic_logger.encode_max_stream_data_frame(
maximum=max_stream_data, stream_id=stream_id
)
)
# check stream direction
self._assert_stream_can_send(frame_type, stream_id)
stream = self._get_or_create_stream(frame_type, stream_id)
if max_stream_data > stream.max_stream_data_remote:
self._logger.debug(
"Stream %d remote max_stream_data raised to %d",
stream_id,
max_stream_data,
)
stream.max_stream_data_remote = max_stream_data
def _handle_max_streams_bidi_frame(
self, context: QuicReceiveContext, frame_type: int, buf: Buffer
) -> None:
"""
Handle a MAX_STREAMS_BIDI frame.
This raises number of bidirectional streams we can initiate to the peer.
"""
max_streams = buf.pull_uint_var()
# log frame
if self._quic_logger is not None:
context.quic_logger_frames.append(
self._quic_logger.encode_connection_limit_frame(
frame_type=frame_type, maximum=max_streams
)
)
if max_streams > self._remote_max_streams_bidi:
self._logger.debug("Remote max_streams_bidi raised to %d", max_streams)
self._remote_max_streams_bidi = max_streams
self._unblock_streams(is_unidirectional=False)
def _handle_max_streams_uni_frame(
self, context: QuicReceiveContext, frame_type: int, buf: Buffer
) -> None:
"""
Handle a MAX_STREAMS_UNI frame.
This raises number of unidirectional streams we can initiate to the peer.
"""
max_streams = buf.pull_uint_var()
# log frame
if self._quic_logger is not None:
context.quic_logger_frames.append(
self._quic_logger.encode_connection_limit_frame(
frame_type=frame_type, maximum=max_streams
)
)
if max_streams > self._remote_max_streams_uni:
self._logger.debug("Remote max_streams_uni raised to %d", max_streams)
self._remote_max_streams_uni = max_streams
self._unblock_streams(is_unidirectional=True)
def _handle_new_connection_id_frame(
self, context: QuicReceiveContext, frame_type: int, buf: Buffer
) -> None:
"""
Handle a NEW_CONNECTION_ID frame.
"""
sequence_number = buf.pull_uint_var()
retire_prior_to = buf.pull_uint_var()
length = buf.pull_uint8()
connection_id = buf.pull_bytes(length)
stateless_reset_token = buf.pull_bytes(16)
#print("PROCESS NEW CID FRAME") #DEBUG*
# log frame
if self._quic_logger is not None:
context.quic_logger_frames.append(
self._quic_logger.encode_new_connection_id_frame(
connection_id=connection_id,
retire_prior_to=retire_prior_to,
sequence_number=sequence_number,
stateless_reset_token=stateless_reset_token,
)
)
# sanity check
if retire_prior_to > sequence_number:
raise QuicConnectionError(
error_code=QuicErrorCode.PROTOCOL_VIOLATION,
frame_type=frame_type,
reason_phrase="retire_prior_to is greater than the sequence_number",
)
# determine which CIDs to retire
change_cid = False
retire = list(
filter(
lambda c: c.sequence_number < retire_prior_to, self._peer_cid_available
)
)
if self._peer_cid.sequence_number < retire_prior_to:
change_cid = True
retire.insert(0, self._peer_cid)
# update available CIDs
self._peer_cid_available = list(
filter(
lambda c: c.sequence_number >= retire_prior_to, self._peer_cid_available
)
)
if sequence_number not in self._peer_cid_sequence_numbers:
self._peer_cid_available.append(
QuicConnectionId(
cid=connection_id,
sequence_number=sequence_number,
stateless_reset_token=stateless_reset_token,
)
)
self._peer_cid_sequence_numbers.add(sequence_number)
# retire previous CIDs
for quic_connection_id in retire:
self._retire_peer_cid(quic_connection_id)
# assign new CID if we retired the active one
if change_cid:
self._consume_peer_cid()
# check number of active connection IDs, including the selected one
if 1 + len(self._peer_cid_available) > self._local_active_connection_id_limit:
raise QuicConnectionError(
error_code=QuicErrorCode.CONNECTION_ID_LIMIT_ERROR,
frame_type=frame_type,
reason_phrase="Too many active connection IDs",
)
def _handle_new_token_frame(
self, context: QuicReceiveContext, frame_type: int, buf: Buffer
) -> None:
"""
Handle a NEW_TOKEN frame.
"""
length = buf.pull_uint_var()
token = buf.pull_bytes(length)
# log frame
if self._quic_logger is not None:
context.quic_logger_frames.append(
self._quic_logger.encode_new_token_frame(token=token)
)
if not self._is_client:
raise QuicConnectionError(
error_code=QuicErrorCode.PROTOCOL_VIOLATION,
frame_type=frame_type,
reason_phrase="Clients must not send NEW_TOKEN frames",
)
def _handle_padding_frame(
self, context: QuicReceiveContext, frame_type: int, buf: Buffer
) -> None:
"""
Handle a PADDING frame.
"""
# consume padding
pos = buf.tell()
for byte in buf.data_slice(pos, buf.capacity):
if byte:
break
pos += 1
buf.seek(pos)
# log frame
if self._quic_logger is not None:
context.quic_logger_frames.append(self._quic_logger.encode_padding_frame())
def _handle_path_challenge_frame(
self, context: QuicReceiveContext, frame_type: int, buf: Buffer
) -> None:
"""
Handle a PATH_CHALLENGE frame.
"""
data = buf.pull_bytes(8)
now = time.time() #CARLO
print("Process PATH CHALLENGE FRAME at: " + str(now)) #DEBUG2* CARLO
# log frame
if self._quic_logger is not None:
context.quic_logger_frames.append(
self._quic_logger.encode_path_challenge_frame(data=data)
)
context.network_path.remote_challenge = data
def _handle_path_response_frame(
self, context: QuicReceiveContext, frame_type: int, buf: Buffer
) -> None:
"""
Handle a PATH_RESPONSE frame.
"""
data = buf.pull_bytes(8)
now = time.time() #CARLO
print("PROCESS PATH RESPONSE FRAME at: " + str(now)) #DEBUG2* CARLO
# log frame
if self._quic_logger is not None:
context.quic_logger_frames.append(
self._quic_logger.encode_path_response_frame(data=data)
)
if data != context.network_path.local_challenge:
raise QuicConnectionError(
error_code=QuicErrorCode.PROTOCOL_VIOLATION,
frame_type=frame_type,
reason_phrase="Response does not match challenge",
)
else: #DEBUG2*
if self._previous_server_address is not None: #DEBUG2*
print("MIGRATION SUCCESSFULLY TERMINATED") #DEBUG2*
self._previous_server_address = None #DEBUG2*
self._server_migration_address = None #DEBUG2*
if self._migration_strategy_fast: #case Explicit - No Loss
self._synched = False
self._logger.debug(
"Network path %s validated by challenge", context.network_path.addr
)
context.network_path.is_validated = True
def _handle_ping_frame(
self, context: QuicReceiveContext, frame_type: int, buf: Buffer
) -> None:
"""
Handle a PING frame.
"""
now = time.time() #CARLO
print("Process PING FRAME at: " + str(now)) #DEBUG2* CARLO
# log frame
if self._quic_logger is not None:
context.quic_logger_frames.append(self._quic_logger.encode_ping_frame())
#DEBUG2*********************
def _handle_server_migration_frame(
self, context: QuicReceiveContext, frame_type: int, buf: Buffer
) -> None:
"""
Handle a SERVER MIGRATION frame.
"""
now = time.time() #CARLO
print("Process SERVER MIGRATION FRAME at: " + str(now)) #DEBUG2* CARLO
#t1 = Thread(target=clientCoAP, args=(self._migration_type,list_addr_server,))
#t1.start()
#Retrieve and convert new ip address of the server
data = buf.pull_uint32(4)
ip = str(ipaddress.IPv4Address(data))
print("IP ADDRESS RECEIVED:")
print(ip)
# log frame
if self._quic_logger is not None:
context.quic_logger_frames.append(
self._quic_logger.encode_server_migration_frame(data=data)
)
#Create QuicNetworkPath
string = '::ffff:'+ip
addr = (string,4433,0,0)
network_path = QuicNetworkPath(addr)
#Adding network path
if network_path not in self._network_paths:
self._network_paths.append(network_path)
#DEBUG V2
#if self._migration_strategy_fast:
#self._change_addr_fast = True
#DEBUG V2
self._previous_server_address = context.network_path
self._server_migration_address = network_path
# log frame
if self._quic_logger is not None:
context.quic_logger_frames.append(self._quic_logger.encode_server_migration_frame(data))
#DEBUG2*********************
#DEBUG2 TEST*********************
def _handle_trigger_frame(
self, context: QuicReceiveContext, frame_type: int, buf: Buffer
) -> None:
"""
Handle a TRIGGER frame.
"""
now = time.time() #CARLO
print("Process TRIGGER FRAME at: " + str(now)) #DEBUG2* CARLO
self._server_triggered_to_migrate = True
# log frame
if self._quic_logger is not None:
context.quic_logger_frames.append(self._quic_logger.encode_trigger_frame())
#DEBUG2 TEST*********************
def _handle_reset_stream_frame(
self, context: QuicReceiveContext, frame_type: int, buf: Buffer
) -> None:
"""
Handle a RESET_STREAM frame.
"""
stream_id = buf.pull_uint_var()
error_code = buf.pull_uint_var()
final_size = buf.pull_uint_var()
# log frame
if self._quic_logger is not None:
context.quic_logger_frames.append(
self._quic_logger.encode_reset_stream_frame(
error_code=error_code, final_size=final_size, stream_id=stream_id
)
)
# check stream direction
self._assert_stream_can_receive(frame_type, stream_id)
# check flow-control limits
stream = self._get_or_create_stream(frame_type, stream_id)
if final_size > stream.max_stream_data_local:
raise QuicConnectionError(
error_code=QuicErrorCode.FLOW_CONTROL_ERROR,
frame_type=frame_type,
reason_phrase="Over stream data limit",
)
newly_received = max(0, final_size - stream._recv_highest)
if self._local_max_data.used + newly_received > self._local_max_data.value:
raise QuicConnectionError(
error_code=QuicErrorCode.FLOW_CONTROL_ERROR,
frame_type=frame_type,
reason_phrase="Over connection data limit",
)
# process reset
self._logger.info(
"Stream %d reset by peer (error code %d, final size %d)",
stream_id,
error_code,
final_size,
)
try:
event = stream.handle_reset(error_code=error_code, final_size=final_size)
except FinalSizeError as exc:
raise QuicConnectionError(
error_code=QuicErrorCode.FINAL_SIZE_ERROR,
frame_type=frame_type,
reason_phrase=str(exc),
)
if event is not None:
self._events.append(event)
self._local_max_data.used += newly_received
def _handle_retire_connection_id_frame(
self, context: QuicReceiveContext, frame_type: int, buf: Buffer
) -> None:
"""
Handle a RETIRE_CONNECTION_ID frame.
"""
sequence_number = buf.pull_uint_var()
#print("PROCESS RETIRE CID FRAME") #DEBUG*
# log frame
if self._quic_logger is not None:
context.quic_logger_frames.append(
self._quic_logger.encode_retire_connection_id_frame(sequence_number)
)
if sequence_number >= self._host_cid_seq:
raise QuicConnectionError(
error_code=QuicErrorCode.PROTOCOL_VIOLATION,
frame_type=frame_type,
reason_phrase="Cannot retire unknown connection ID",
)
# find the connection ID by sequence number
for index, connection_id in enumerate(self._host_cids):
if connection_id.sequence_number == sequence_number:
if connection_id.cid == context.host_cid:
raise QuicConnectionError(
error_code=QuicErrorCode.PROTOCOL_VIOLATION,
frame_type=frame_type,
reason_phrase="Cannot retire current connection ID",
)
self._logger.debug(
"Peer retiring CID %s (%d)",
dump_cid(connection_id.cid),
connection_id.sequence_number,
)
del self._host_cids[index]
self._events.append(
events.ConnectionIdRetired(connection_id=connection_id.cid)
)
break
# issue a new connection ID
self._replenish_connection_ids()
def _handle_stop_sending_frame(
self, context: QuicReceiveContext, frame_type: int, buf: Buffer
) -> None:
"""
Handle a STOP_SENDING frame.
"""
stream_id = buf.pull_uint_var()
error_code = buf.pull_uint_var() # application error code
# log frame
if self._quic_logger is not None:
context.quic_logger_frames.append(
self._quic_logger.encode_stop_sending_frame(
error_code=error_code, stream_id=stream_id
)
)
# check stream direction
self._assert_stream_can_send(frame_type, stream_id)
self._get_or_create_stream(frame_type, stream_id)
def _handle_stream_frame(
self, context: QuicReceiveContext, frame_type: int, buf: Buffer
) -> None:
"""
Handle a STREAM frame.
"""
stream_id = buf.pull_uint_var()
if frame_type & 4:
offset = buf.pull_uint_var()
else:
offset = 0
if frame_type & 2:
length = buf.pull_uint_var()
else:
length = buf.capacity - buf.tell()
if offset + length > UINT_VAR_MAX:
raise QuicConnectionError(
error_code=QuicErrorCode.FRAME_ENCODING_ERROR,
frame_type=frame_type,
reason_phrase="offset + length cannot exceed 2^62 - 1",
)
frame = QuicStreamFrame(
offset=offset, data=buf.pull_bytes(length), fin=bool(frame_type & 1)
)
# log frame
if self._quic_logger is not None:
context.quic_logger_frames.append(
self._quic_logger.encode_stream_frame(frame, stream_id=stream_id)
)
# check stream direction
self._assert_stream_can_receive(frame_type, stream_id)
# check flow-control limits
stream = self._get_or_create_stream(frame_type, stream_id)
if offset + length > stream.max_stream_data_local:
raise QuicConnectionError(
error_code=QuicErrorCode.FLOW_CONTROL_ERROR,
frame_type=frame_type,
reason_phrase="Over stream data limit",
)
newly_received = max(0, offset + length - stream._recv_highest)
if self._local_max_data.used + newly_received > self._local_max_data.value:
raise QuicConnectionError(
error_code=QuicErrorCode.FLOW_CONTROL_ERROR,
frame_type=frame_type,
reason_phrase="Over connection data limit",
)
# process data
try:
event = stream.add_frame(frame)
except FinalSizeError as exc:
raise QuicConnectionError(
error_code=QuicErrorCode.FINAL_SIZE_ERROR,
frame_type=frame_type,
reason_phrase=str(exc),
)
if event is not None:
self._events.append(event)
self._local_max_data.used += newly_received
def _handle_stream_data_blocked_frame(
self, context: QuicReceiveContext, frame_type: int, buf: Buffer
) -> None:
"""
Handle a STREAM_DATA_BLOCKED frame.
"""
stream_id = buf.pull_uint_var()
limit = buf.pull_uint_var()
# log frame
if self._quic_logger is not None:
context.quic_logger_frames.append(
self._quic_logger.encode_stream_data_blocked_frame(
limit=limit, stream_id=stream_id
)
)
# check stream direction
self._assert_stream_can_receive(frame_type, stream_id)
self._get_or_create_stream(frame_type, stream_id)
def _handle_streams_blocked_frame(
self, context: QuicReceiveContext, frame_type: int, buf: Buffer
) -> None:
"""
Handle a STREAMS_BLOCKED frame.
"""
limit = buf.pull_uint_var()
# log frame
if self._quic_logger is not None:
context.quic_logger_frames.append(
self._quic_logger.encode_streams_blocked_frame(
is_unidirectional=frame_type == QuicFrameType.STREAMS_BLOCKED_UNI,
limit=limit,
)
)
def _log_key_retired(self, key_type: str, trigger: str) -> None:
"""
Log a key retirement.
"""
if self._quic_logger is not None:
self._quic_logger.log_event(
category="security",
event="key_retired",
data={"key_type": key_type, "trigger": trigger},
)
def _log_key_updated(self, key_type: str, trigger: str) -> None:
"""
Log a key update.
"""
if self._quic_logger is not None:
self._quic_logger.log_event(
category="security",
event="key_updated",
data={"key_type": key_type, "trigger": trigger},
)
def _on_ack_delivery(
self, delivery: QuicDeliveryState, space: QuicPacketSpace, highest_acked: int
) -> None:
"""
Callback when an ACK frame is acknowledged or lost.
"""
if delivery == QuicDeliveryState.ACKED:
space.ack_queue.subtract(0, highest_acked + 1)
def _on_connection_limit_delivery(
self, delivery: QuicDeliveryState, limit: Limit
) -> None:
"""
Callback when a MAX_DATA or MAX_STREAMS frame is acknowledged or lost.
"""
if delivery != QuicDeliveryState.ACKED:
limit.sent = 0
def _on_handshake_done_delivery(self, delivery: QuicDeliveryState) -> None:
"""
Callback when a HANDSHAKE_DONE frame is acknowledged or lost.
"""
if delivery != QuicDeliveryState.ACKED:
self._handshake_done_pending = True
def _on_max_stream_data_delivery(
self, delivery: QuicDeliveryState, stream: QuicStream
) -> None:
"""
Callback when a MAX_STREAM_DATA frame is acknowledged or lost.
"""
if delivery != QuicDeliveryState.ACKED:
stream.max_stream_data_local_sent = 0
def _on_new_connection_id_delivery(
self, delivery: QuicDeliveryState, connection_id: QuicConnectionId
) -> None:
"""
Callback when a NEW_CONNECTION_ID frame is acknowledged or lost.
"""
if delivery != QuicDeliveryState.ACKED:
connection_id.was_sent = False
def _on_ping_delivery(
self, delivery: QuicDeliveryState, uids: Sequence[int]
) -> None:
"""
Callback when a PING frame is acknowledged or lost.
"""
if delivery == QuicDeliveryState.ACKED:
self._logger.debug("Received PING%s response", "" if uids else " (probe)")
for uid in uids:
self._events.append(events.PingAcknowledged(uid=uid))
else:
self._ping_pending.extend(uids)
#DEBUG2************************ #PERF EV AUTOMATION V2****** #DEBUG V3******
def _on_server_migration_delivery(
self, delivery: QuicDeliveryState
) -> None:
"""
Callback when a SERVER MIGRATION frame is acknowledged or lost.
"""
if delivery != QuicDeliveryState.ACKED:
print("SERVER MIGRATION FRAME WAS LOST")
self._server_triggered_to_migrate = True
else:
print("SERVER MIGRATION FRAME WAS ACKNOWLEDGED")
list_addr_server = self._list_addr_server
#PERF EV V2 AUTOMATION Client CoAP****
t1 = Thread(target=clientCoAP, args=(self._migration_type,list_addr_server,))
t1.start()
#PERF EV V2 AUTOMATION Client CoAP*****
self._list_addr_server = list(reversed(self._list_addr_server))
#DEBUG2************************ #PERF EV AUTOMATION V2****** #DEBUG V3******
#DEBUG2 TEST************************
def _on_trigger_delivery(
self, delivery: QuicDeliveryState
) -> None:
"""
Callback when a TRIGGER frame is acknowledged or lost.
"""
if delivery != QuicDeliveryState.ACKED:
self._trigger_period = True
#DEBUG2 TEST************************
def _on_retire_connection_id_delivery(
self, delivery: QuicDeliveryState, sequence_number: int
) -> None:
"""
Callback when a RETIRE_CONNECTION_ID frame is acknowledged or lost.
"""
if delivery != QuicDeliveryState.ACKED:
self._retire_connection_ids.append(sequence_number)
def _payload_received(
self, context: QuicReceiveContext, plain: bytes
) -> Tuple[bool, bool]:
"""
Handle a QUIC packet payload.
"""
buf = Buffer(data=plain)
is_ack_eliciting = False
is_probing = None
cont=0 #DEBUG*
while not buf.eof():
frame_type = buf.pull_uint_var()
#print("FRAME TYPE: " + str(frame_type)) #DEBUG*
cont+=1 #DEBUG*
# check frame type is known
try:
frame_handler, frame_epochs = self.__frame_handlers[frame_type]
except KeyError:
raise QuicConnectionError(
error_code=QuicErrorCode.PROTOCOL_VIOLATION,
frame_type=frame_type,
reason_phrase="Unknown frame type",
)
# check frame is allowed for the epoch
if context.epoch not in frame_epochs:
raise QuicConnectionError(
error_code=QuicErrorCode.PROTOCOL_VIOLATION,
frame_type=frame_type,
reason_phrase="Unexpected frame type",
)
# handle the frame
try:
frame_handler(context, frame_type, buf)
except BufferReadError:
raise QuicConnectionError(
error_code=QuicErrorCode.FRAME_ENCODING_ERROR,
frame_type=frame_type,
reason_phrase="Failed to parse frame",
)
# update ACK only / probing flags
if frame_type not in NON_ACK_ELICITING_FRAME_TYPES:
is_ack_eliciting = True
if frame_type not in PROBING_FRAME_TYPES:
is_probing = False
elif is_probing is None:
is_probing = True
#print("NUMBER OF FRAMES PROCESSED:" + str(cont)) #DEBUG*
return is_ack_eliciting, bool(is_probing)
def _replenish_connection_ids(self) -> None:
"""
Generate new connection IDs.
"""
while len(self._host_cids) < min(8, self._remote_active_connection_id_limit):
self._host_cids.append(
QuicConnectionId(
cid=os.urandom(self._configuration.connection_id_length),
sequence_number=self._host_cid_seq,
stateless_reset_token=os.urandom(16),
)
)
self._host_cid_seq += 1
def _retire_peer_cid(self, connection_id: QuicConnectionId) -> None:
"""
Retire a destination connection ID.
"""
self._logger.debug(
"Retiring CID %s (%d)",
dump_cid(connection_id.cid),
connection_id.sequence_number,
)
self._retire_connection_ids.append(connection_id.sequence_number)
def _push_crypto_data(self) -> None:
for epoch, buf in self._crypto_buffers.items():
self._crypto_streams[epoch].write(buf.data)
buf.seek(0)
def _send_probe(self) -> None:
self._probe_pending = True
def _parse_transport_parameters(
self, data: bytes, from_session_ticket: bool = False
) -> None:
"""
Parse and apply remote transport parameters.
`from_session_ticket` is `True` when restoring saved transport parameters,
and `False` when handling received transport parameters.
"""
quic_transport_parameters = pull_quic_transport_parameters(Buffer(data=data))
# log event
if self._quic_logger is not None and not from_session_ticket:
self._quic_logger.log_event(
category="transport",
event="parameters_set",
data=self._quic_logger.encode_transport_parameters(
owner="remote", parameters=quic_transport_parameters
),
)
# validate remote parameters
if not self._is_client:
for attr in [
"original_destination_connection_id",
"preferred_address",
"retry_source_connection_id",
"stateless_reset_token",
]:
if getattr(quic_transport_parameters, attr) is not None:
raise QuicConnectionError(
error_code=QuicErrorCode.TRANSPORT_PARAMETER_ERROR,
frame_type=QuicFrameType.CRYPTO,
reason_phrase="%s is not allowed for clients" % attr,
)
if not from_session_ticket:
if self._is_client and (
quic_transport_parameters.original_destination_connection_id
!= self._original_destination_connection_id
):
raise QuicConnectionError(
error_code=QuicErrorCode.TRANSPORT_PARAMETER_ERROR,
frame_type=QuicFrameType.CRYPTO,
reason_phrase="original_destination_connection_id does not match",
)
if self._is_client and (
quic_transport_parameters.retry_source_connection_id
!= self._retry_source_connection_id
):
raise QuicConnectionError(
error_code=QuicErrorCode.TRANSPORT_PARAMETER_ERROR,
frame_type=QuicFrameType.CRYPTO,
reason_phrase="retry_source_connection_id does not match",
)
if (
quic_transport_parameters.active_connection_id_limit is not None
and quic_transport_parameters.active_connection_id_limit < 2
):
raise QuicConnectionError(
error_code=QuicErrorCode.TRANSPORT_PARAMETER_ERROR,
frame_type=QuicFrameType.CRYPTO,
reason_phrase="active_connection_id_limit must be no less than 2",
)
if (
self._is_client
and self._peer_cid.sequence_number == 0
and quic_transport_parameters.stateless_reset_token is not None
):
self._peer_cid.stateless_reset_token = (
quic_transport_parameters.stateless_reset_token
)
# store remote parameters
if not from_session_ticket:
if quic_transport_parameters.ack_delay_exponent is not None:
self._remote_ack_delay_exponent = self._remote_ack_delay_exponent
if quic_transport_parameters.max_ack_delay is not None:
self._loss.max_ack_delay = (
quic_transport_parameters.max_ack_delay / 1000.0
)
if quic_transport_parameters.active_connection_id_limit is not None:
self._remote_active_connection_id_limit = (
quic_transport_parameters.active_connection_id_limit
)
if quic_transport_parameters.max_idle_timeout is not None:
self._remote_max_idle_timeout = (
quic_transport_parameters.max_idle_timeout / 1000.0
)
self._remote_max_datagram_frame_size = (
quic_transport_parameters.max_datagram_frame_size
)
for param in [
"max_data",
"max_stream_data_bidi_local",
"max_stream_data_bidi_remote",
"max_stream_data_uni",
"max_streams_bidi",
"max_streams_uni",
]:
value = getattr(quic_transport_parameters, "initial_" + param)
if value is not None:
setattr(self, "_remote_" + param, value)
def _serialize_transport_parameters(self) -> bytes:
quic_transport_parameters = QuicTransportParameters(
ack_delay_exponent=self._local_ack_delay_exponent,
active_connection_id_limit=self._local_active_connection_id_limit,
max_idle_timeout=int(self._configuration.idle_timeout * 1000),
initial_max_data=self._local_max_data.value,
initial_max_stream_data_bidi_local=self._local_max_stream_data_bidi_local,
initial_max_stream_data_bidi_remote=self._local_max_stream_data_bidi_remote,
initial_max_stream_data_uni=self._local_max_stream_data_uni,
initial_max_streams_bidi=self._local_max_streams_bidi.value,
initial_max_streams_uni=self._local_max_streams_uni.value,
initial_source_connection_id=self._local_initial_source_connection_id,
max_ack_delay=25,
max_datagram_frame_size=self._configuration.max_datagram_frame_size,
quantum_readiness=b"Q" * 1200
if self._configuration.quantum_readiness_test
else None,
stateless_reset_token=self._host_cids[0].stateless_reset_token,
)
if not self._is_client and (
self._version >= QuicProtocolVersion.DRAFT_28
or self._retry_source_connection_id
):
quic_transport_parameters.original_destination_connection_id = (
self._original_destination_connection_id
)
quic_transport_parameters.retry_source_connection_id = (
self._retry_source_connection_id
)
# log event
if self._quic_logger is not None:
self._quic_logger.log_event(
category="transport",
event="parameters_set",
data=self._quic_logger.encode_transport_parameters(
owner="local", parameters=quic_transport_parameters
),
)
buf = Buffer(capacity=3 * PACKET_MAX_SIZE)
push_quic_transport_parameters(buf, quic_transport_parameters)
return buf.data
def _set_state(self, state: QuicConnectionState) -> None:
self._logger.debug("%s -> %s", self._state, state)
self._state = state
def _stream_can_receive(self, stream_id: int) -> bool:
return stream_is_client_initiated(
stream_id
) != self._is_client or not stream_is_unidirectional(stream_id)
def _stream_can_send(self, stream_id: int) -> bool:
return stream_is_client_initiated(
stream_id
) == self._is_client or not stream_is_unidirectional(stream_id)
def _unblock_streams(self, is_unidirectional: bool) -> None:
if is_unidirectional:
max_stream_data_remote = self._remote_max_stream_data_uni
max_streams = self._remote_max_streams_uni
streams_blocked = self._streams_blocked_uni
else:
max_stream_data_remote = self._remote_max_stream_data_bidi_remote
max_streams = self._remote_max_streams_bidi
streams_blocked = self._streams_blocked_bidi
while streams_blocked and streams_blocked[0].stream_id // 4 < max_streams:
stream = streams_blocked.pop(0)
stream.is_blocked = False
stream.max_stream_data_remote = max_stream_data_remote
if not self._streams_blocked_bidi and not self._streams_blocked_uni:
self._streams_blocked_pending = False
def _update_traffic_key(
self,
direction: tls.Direction,
epoch: tls.Epoch,
cipher_suite: tls.CipherSuite,
secret: bytes,
) -> None:
"""
Callback which is invoked by the TLS engine when new traffic keys are
available.
"""
secrets_log_file = self._configuration.secrets_log_file
if secrets_log_file is not None:
label_row = self._is_client == (direction == tls.Direction.DECRYPT)
label = SECRETS_LABELS[label_row][epoch.value]
secrets_log_file.write(
"%s %s %s\n" % (label, self.tls.client_random.hex(), secret.hex())
)
secrets_log_file.flush()
crypto = self._cryptos[epoch]
if direction == tls.Direction.ENCRYPT:
crypto.send.setup(
cipher_suite=cipher_suite, secret=secret, version=self._version
)
else:
crypto.recv.setup(
cipher_suite=cipher_suite, secret=secret, version=self._version
)
def _write_application(
self, builder: QuicPacketBuilder, network_path: QuicNetworkPath, now: float, counter: int, hmstrategy: int, n_request_migration:int, interval_migration:int #DEBUG2 TEST DEBUG V2* PERF EV AUTOMATION* DEBUG V3*
) -> None:
crypto_stream: Optional[QuicStream] = None
if self._cryptos[tls.Epoch.ONE_RTT].send.is_valid():
crypto = self._cryptos[tls.Epoch.ONE_RTT]
crypto_stream = self._crypto_streams[tls.Epoch.ONE_RTT]
packet_type = PACKET_TYPE_ONE_RTT
elif self._cryptos[tls.Epoch.ZERO_RTT].send.is_valid():
crypto = self._cryptos[tls.Epoch.ZERO_RTT]
packet_type = PACKET_TYPE_ZERO_RTT
else:
return
space = self._spaces[tls.Epoch.ONE_RTT]
while True:
# apply pacing, except if we have ACKs to send
if space.ack_at is None or space.ack_at >= now:
self._pacing_at = self._loss._pacer.next_send_time(now=now)
if self._pacing_at is not None:
break
builder.start_packet(packet_type, crypto)
if self._handshake_complete:
# ACK
if space.ack_at is not None and space.ack_at <= now:
self._write_ack_frame(builder=builder, space=space, now=now)
# HANDSHAKE_DONE
if self._handshake_done_pending:
self._write_handshake_done_frame(builder=builder)
self._handshake_done_pending = False
# PATH CHALLENGE
if (
not network_path.is_validated
and network_path.local_challenge is None
):
challenge = os.urandom(8)
self._write_path_challenge_frame(
builder=builder, challenge=challenge
)
network_path.local_challenge = challenge
# PATH RESPONSE
if network_path.remote_challenge is not None:
self._write_path_response_frame(
builder=builder, challenge=network_path.remote_challenge
)
network_path.remote_challenge = None
# NEW_CONNECTION_ID
for connection_id in self._host_cids:
if not connection_id.was_sent:
self._write_new_connection_id_frame(
builder=builder, connection_id=connection_id
)
# RETIRE_CONNECTION_ID
for sequence_number in self._retire_connection_ids[:]:
self._write_retire_connection_id_frame(
builder=builder, sequence_number=sequence_number
)
self._retire_connection_ids.pop(0)
# STREAMS_BLOCKED
if self._streams_blocked_pending:
if self._streams_blocked_bidi:
self._write_streams_blocked_frame(
builder=builder,
frame_type=QuicFrameType.STREAMS_BLOCKED_BIDI,
limit=self._remote_max_streams_bidi,
)
if self._streams_blocked_uni:
self._write_streams_blocked_frame(
builder=builder,
frame_type=QuicFrameType.STREAMS_BLOCKED_UNI,
limit=self._remote_max_streams_uni,
)
self._streams_blocked_pending = False
# MAX_DATA and MAX_STREAMS
self._write_connection_limits(builder=builder, space=space)
#DEBUG V2*********************
#Type of migration strategy #Update FAST VERSION
if hmstrategy == 1 and not self._migration_strategy_fast:
print("SETTO FLAG MIGRATION STRATEGY FAST")
self._migration_strategy_fast = True
#DEBUG V2********************* #Update FAST VERSION
#DEBUG V3*********************
if self._is_client and self._n_request_migration == -1 and self._interval_migration == -1 and n_request_migration != 0 and interval_migration != 0:
self._n_request_migration = n_request_migration
self._interval_migration = interval_migration
#Condition for triggering
if self._is_client and counter == self._n_request_migration: #and self._first_time_trigger:
if not self._trigger_period:
self._trigger_period = True
self._n_request_migration = self._n_request_migration + self._interval_migration
#self._first_time_trigger = False
#DEBUG V3*********************
#TRIGGER
if self._is_client and self._trigger_period:
self._write_trigger_frame(builder=builder)
#DEBUG2 TEST*********************
#DEBUG2*********************
#SERVER MIGRATION
if not self._is_client and self._server_triggered_to_migrate:
self._write_server_migration_frame(builder=builder) #For test case ip address of server inside function
#DEBUG2*********************
# stream-level limits
for stream in self._streams.values():
self._write_stream_limits(builder=builder, space=space, stream=stream)
# PING (user-request)
if self._ping_pending:
self._write_ping_frame(builder, self._ping_pending)
self._ping_pending.clear()
# PING (probe)
if self._probe_pending:
self._write_ping_frame(builder, comment="probe")
self._probe_pending = False
# CRYPTO
if crypto_stream is not None and not crypto_stream.send_buffer_is_empty:
self._write_crypto_frame(
builder=builder, space=space, stream=crypto_stream
)
# DATAGRAM
while self._datagrams_pending:
try:
self._write_datagram_frame(
builder=builder,
data=self._datagrams_pending[0],
frame_type=QuicFrameType.DATAGRAM_WITH_LENGTH,
)
self._datagrams_pending.popleft()
except QuicPacketBuilderStop:
break
# STREAM and RESET_STREAM
for stream in self._streams.values():
if stream.reset_pending:
self._write_reset_stream_frame(
builder=builder,
frame_type=QuicFrameType.RESET_STREAM,
stream=stream,
)
elif not stream.is_blocked and not stream.send_buffer_is_empty:
self._remote_max_data_used += self._write_stream_frame(
builder=builder,
space=space,
stream=stream,
max_offset=min(
stream._send_highest
+ self._remote_max_data
- self._remote_max_data_used,
stream.max_stream_data_remote,
),
)
if builder.packet_is_empty:
break
else:
self._loss._pacer.update_after_send(now=now)
def _write_handshake(
self, builder: QuicPacketBuilder, epoch: tls.Epoch, now: float
) -> None:
crypto = self._cryptos[epoch]
if not crypto.send.is_valid():
return
crypto_stream = self._crypto_streams[epoch]
space = self._spaces[epoch]
while True:
if epoch == tls.Epoch.INITIAL:
packet_type = PACKET_TYPE_INITIAL
else:
packet_type = PACKET_TYPE_HANDSHAKE
builder.start_packet(packet_type, crypto)
# ACK
if space.ack_at is not None:
self._write_ack_frame(builder=builder, space=space, now=now)
# CRYPTO
if not crypto_stream.send_buffer_is_empty:
if self._write_crypto_frame(
builder=builder, space=space, stream=crypto_stream
):
self._probe_pending = False
# PING (probe)
if (
self._probe_pending
and not self._handshake_complete
and (
epoch == tls.Epoch.HANDSHAKE
or not self._cryptos[tls.Epoch.HANDSHAKE].send.is_valid()
)
):
self._write_ping_frame(builder, comment="probe")
self._probe_pending = False
if builder.packet_is_empty:
break
def _write_ack_frame(
self, builder: QuicPacketBuilder, space: QuicPacketSpace, now: float
) -> None:
# calculate ACK delay
ack_delay = now - space.largest_received_time
ack_delay_encoded = int(ack_delay * 1000000) >> self._local_ack_delay_exponent
#print("WRITE ACK FRAME") #DEBUG*
buf = builder.start_frame(
QuicFrameType.ACK,
capacity=ACK_FRAME_CAPACITY,
handler=self._on_ack_delivery,
handler_args=(space, space.largest_received_packet),
)
ranges = push_ack_frame(buf, space.ack_queue, ack_delay_encoded)
space.ack_at = None
# log frame
if self._quic_logger is not None:
builder.quic_logger_frames.append(
self._quic_logger.encode_ack_frame(
ranges=space.ack_queue, delay=ack_delay
)
)
# check if we need to trigger an ACK-of-ACK
if ranges > 1 and builder.packet_number % 8 == 0:
self._write_ping_frame(builder, comment="ACK-of-ACK trigger")
def _write_connection_close_frame(
self,
builder: QuicPacketBuilder,
epoch: tls.Epoch,
error_code: int,
frame_type: Optional[int],
reason_phrase: str,
) -> None:
# convert application-level close to transport-level close in early stages
if frame_type is None and epoch in (tls.Epoch.INITIAL, tls.Epoch.HANDSHAKE):
error_code = QuicErrorCode.APPLICATION_ERROR
frame_type = QuicFrameType.PADDING
reason_phrase = ""
reason_bytes = reason_phrase.encode("utf8")
reason_length = len(reason_bytes)
print("WRITE CONNECTION CLOSE FRAME")
if frame_type is None:
buf = builder.start_frame(
QuicFrameType.APPLICATION_CLOSE,
capacity=APPLICATION_CLOSE_FRAME_CAPACITY + reason_length,
)
buf.push_uint_var(error_code)
buf.push_uint_var(reason_length)
buf.push_bytes(reason_bytes)
else:
buf = builder.start_frame(
QuicFrameType.TRANSPORT_CLOSE,
capacity=TRANSPORT_CLOSE_FRAME_CAPACITY + reason_length,
)
buf.push_uint_var(error_code)
buf.push_uint_var(frame_type)
buf.push_uint_var(reason_length)
buf.push_bytes(reason_bytes)
# log frame
if self._quic_logger is not None:
builder.quic_logger_frames.append(
self._quic_logger.encode_connection_close_frame(
error_code=error_code,
frame_type=frame_type,
reason_phrase=reason_phrase,
)
)
def _write_connection_limits(
self, builder: QuicPacketBuilder, space: QuicPacketSpace
) -> None:
"""
Raise MAX_DATA or MAX_STREAMS if needed.
"""
for limit in (
self._local_max_data,
self._local_max_streams_bidi,
self._local_max_streams_uni,
):
if limit.used * 2 > limit.value:
limit.value *= 2
self._logger.debug("Local %s raised to %d", limit.name, limit.value)
if limit.value != limit.sent:
#print("WRITE CONNECTION'S LIMIT FRAME") #DEBUG*
buf = builder.start_frame(
limit.frame_type,
capacity=CONNECTION_LIMIT_FRAME_CAPACITY,
handler=self._on_connection_limit_delivery,
handler_args=(limit,),
)
buf.push_uint_var(limit.value)
limit.sent = limit.value
# log frame
if self._quic_logger is not None:
builder.quic_logger_frames.append(
self._quic_logger.encode_connection_limit_frame(
frame_type=limit.frame_type, maximum=limit.value,
)
)
def _write_crypto_frame(
self, builder: QuicPacketBuilder, space: QuicPacketSpace, stream: QuicStream
) -> bool:
frame_overhead = 3 + size_uint_var(stream.next_send_offset)
frame = stream.get_frame(builder.remaining_flight_space - frame_overhead)
if frame is not None:
#print("WRITE CRYPTO FRAME") #DEBUG*
buf = builder.start_frame(
QuicFrameType.CRYPTO,
capacity=frame_overhead,
handler=stream.on_data_delivery,
handler_args=(frame.offset, frame.offset + len(frame.data)),
)
buf.push_uint_var(frame.offset)
buf.push_uint16(len(frame.data) | 0x4000)
buf.push_bytes(frame.data)
# log frame
if self._quic_logger is not None:
builder.quic_logger_frames.append(
self._quic_logger.encode_crypto_frame(frame)
)
return True
return False
def _write_datagram_frame(
self, builder: QuicPacketBuilder, data: bytes, frame_type: QuicFrameType
) -> bool:
"""
Write a DATAGRAM frame.
Returns True if the frame was processed, False otherwise.
"""
assert frame_type == QuicFrameType.DATAGRAM_WITH_LENGTH
length = len(data)
frame_size = 1 + size_uint_var(length) + length
#print("WRITE DATAGRAM FRAME") #DEBUG*
buf = builder.start_frame(frame_type, capacity=frame_size)
buf.push_uint_var(length)
buf.push_bytes(data)
# log frame
if self._quic_logger is not None:
builder.quic_logger_frames.append(
self._quic_logger.encode_datagram_frame(length=length)
)
return True
def _write_handshake_done_frame(self, builder: QuicPacketBuilder) -> None:
#print("WRITE HANDSHAKE DONE FRAME") #DEBUG*
builder.start_frame(
QuicFrameType.HANDSHAKE_DONE,
capacity=HANDSHAKE_DONE_FRAME_CAPACITY,
handler=self._on_handshake_done_delivery,
)
# log frame
if self._quic_logger is not None:
builder.quic_logger_frames.append(
self._quic_logger.encode_handshake_done_frame()
)
def _write_new_connection_id_frame(
self, builder: QuicPacketBuilder, connection_id: QuicConnectionId
) -> None:
retire_prior_to = 0 # FIXME
#print("WRITE NEW CID FRAME") #DEBUG*
buf = builder.start_frame(
QuicFrameType.NEW_CONNECTION_ID,
capacity=NEW_CONNECTION_ID_FRAME_CAPACITY,
handler=self._on_new_connection_id_delivery,
handler_args=(connection_id,),
)
buf.push_uint_var(connection_id.sequence_number)
buf.push_uint_var(retire_prior_to)
buf.push_uint8(len(connection_id.cid))
buf.push_bytes(connection_id.cid)
buf.push_bytes(connection_id.stateless_reset_token)
connection_id.was_sent = True
self._events.append(events.ConnectionIdIssued(connection_id=connection_id.cid))
# log frame
if self._quic_logger is not None:
builder.quic_logger_frames.append(
self._quic_logger.encode_new_connection_id_frame(
connection_id=connection_id.cid,
retire_prior_to=retire_prior_to,
sequence_number=connection_id.sequence_number,
stateless_reset_token=connection_id.stateless_reset_token,
)
)
def _write_path_challenge_frame(
self, builder: QuicPacketBuilder, challenge: bytes
) -> None:
buf = builder.start_frame(
QuicFrameType.PATH_CHALLENGE, capacity=PATH_CHALLENGE_FRAME_CAPACITY
)
buf.push_bytes(challenge)
now = time.time() #CARLO
print("WRITE PATH CHALLENGE FRAME at: " + str(now)) #DEBUG2* CARLO
# log frame
if self._quic_logger is not None:
builder.quic_logger_frames.append(
self._quic_logger.encode_path_challenge_frame(data=challenge)
)
def _write_path_response_frame(
self, builder: QuicPacketBuilder, challenge: bytes
) -> None:
buf = builder.start_frame(
QuicFrameType.PATH_RESPONSE, capacity=PATH_RESPONSE_FRAME_CAPACITY
)
buf.push_bytes(challenge)
now = time.time() #CARLO
print("WRITE PATH RESPONSE FRAME at: " + str(now)) #DEBUG2* CARLO
# log frame
if self._quic_logger is not None:
builder.quic_logger_frames.append(
self._quic_logger.encode_path_response_frame(data=challenge)
)
def _write_ping_frame(
self, builder: QuicPacketBuilder, uids: List[int] = [], comment=""
):
now = time.time() #CARLO
print("WRITE PING FRAME at: " + str(now)) #DEBUG2* CARLO
builder.start_frame(
QuicFrameType.PING,
capacity=PING_FRAME_CAPACITY,
handler=self._on_ping_delivery,
handler_args=(tuple(uids),),
)
self._logger.debug(
"Sending PING%s in packet %d",
" (%s)" % comment if comment else "",
builder.packet_number,
)
# log frame
if self._quic_logger is not None:
builder.quic_logger_frames.append(self._quic_logger.encode_ping_frame())
#DEBUG2*********************
def _write_server_migration_frame(
self, builder: QuicPacketBuilder
):
now = time.time() #CARLO
print("WRITE SERVER MIGRATION FRAME at: " + str(now)) #DEBUG2* CARLO
ip = self._list_addr_server[0] #"172.16.4.232" #192.168.178.55 #DEBUG V3*
ip_int = int(ipaddress.IPv4Address(ip))
print("IP ADDRESS SENT:")
print(ip)
buf = builder.start_frame(
QuicFrameType.SERVER_MIGRATION,
capacity=SERVER_MIGRATION_FRAME_CAPACITY,
handler=self._on_server_migration_delivery,
)
buf.push_uint32(ip_int)
self._server_triggered_to_migrate = False
self._logger.debug(
"Sending SERVER MIGRATION in packet %d",
builder.packet_number,
)
# log frame
if self._quic_logger is not None:
builder.quic_logger_frames.append(self._quic_logger.encode_server_migration_frame(data=ip_int))
#DEBUG2*********************
#DEBUG2 TEST*********************
def _write_trigger_frame(
self, builder: QuicPacketBuilder
):
now = time.time() #CARLO
print("WRITE TRIGGER FRAME at: " + str(now)) #DEBUG2* CARLO
builder.start_frame(
QuicFrameType.TRIGGER,
capacity=PING_FRAME_CAPACITY, #Leave ping frame capacity because I don't need to send data
handler = self._on_trigger_delivery,
)
self._trigger_period = False
self._logger.debug(
"Sending TRIGGER in packet %d",
builder.packet_number,
)
# log frame
if self._quic_logger is not None:
builder.quic_logger_frames.append(self._quic_logger.encode_trigger_frame())
#DEBUG2 TEST*********************
def _write_reset_stream_frame(
self, builder: QuicPacketBuilder, frame_type: QuicFrameType, stream: QuicStream,
) -> None:
#print("WRITE RESET STREAM FRAME") #DEBUG*
buf = builder.start_frame(
frame_type=frame_type,
capacity=RESET_STREAM_CAPACITY,
handler=stream.on_reset_delivery,
)
reset = stream.get_reset_frame()
buf.push_uint_var(stream.stream_id)
buf.push_uint_var(reset.error_code)
buf.push_uint_var(reset.final_size)
# log frame
if self._quic_logger is not None:
builder.quic_logger_frames.append(
self._quic_logger.encode_reset_stream_frame(
error_code=reset.error_code,
final_size=reset.final_size,
stream_id=stream.stream_id,
)
)
def _write_retire_connection_id_frame(
self, builder: QuicPacketBuilder, sequence_number: int
) -> None:
buf = builder.start_frame(
QuicFrameType.RETIRE_CONNECTION_ID,
capacity=RETIRE_CONNECTION_ID_CAPACITY,
handler=self._on_retire_connection_id_delivery,
handler_args=(sequence_number,),
)
buf.push_uint_var(sequence_number)
#print("WRITE RETIRE CID FRAME") #DEBUG*
# log frame
if self._quic_logger is not None:
builder.quic_logger_frames.append(
self._quic_logger.encode_retire_connection_id_frame(sequence_number)
)
def _write_stream_frame(
self,
builder: QuicPacketBuilder,
space: QuicPacketSpace,
stream: QuicStream,
max_offset: int,
) -> int:
# the frame data size is constrained by our peer's MAX_DATA and
# the space available in the current packet
frame_overhead = (
3
+ size_uint_var(stream.stream_id)
+ (size_uint_var(stream.next_send_offset) if stream.next_send_offset else 0)
)
previous_send_highest = stream._send_highest
frame = stream.get_frame(
builder.remaining_flight_space - frame_overhead, max_offset
)
if frame is not None:
frame_type = QuicFrameType.STREAM_BASE | 2 # length
if frame.offset:
frame_type |= 4
if frame.fin:
frame_type |= 1
#print("WRITE STREAM FRAME") #DEBUG*
buf = builder.start_frame(
frame_type,
capacity=frame_overhead,
handler=stream.on_data_delivery,
handler_args=(frame.offset, frame.offset + len(frame.data)),
)
buf.push_uint_var(stream.stream_id)
if frame.offset:
buf.push_uint_var(frame.offset)
buf.push_uint16(len(frame.data) | 0x4000)
buf.push_bytes(frame.data)
# log frame
if self._quic_logger is not None:
builder.quic_logger_frames.append(
self._quic_logger.encode_stream_frame(
frame, stream_id=stream.stream_id
)
)
return stream._send_highest - previous_send_highest
else:
return 0
def _write_stream_limits(
self, builder: QuicPacketBuilder, space: QuicPacketSpace, stream: QuicStream
) -> None:
"""
Raise MAX_STREAM_DATA if needed.
The only case where `stream.max_stream_data_local` is zero is for
locally created unidirectional streams. We skip such streams to avoid
spurious logging.
"""
if (
stream.max_stream_data_local
and stream._recv_highest * 2 > stream.max_stream_data_local
):
stream.max_stream_data_local *= 2
self._logger.debug(
"Stream %d local max_stream_data raised to %d",
stream.stream_id,
stream.max_stream_data_local,
)
if stream.max_stream_data_local_sent != stream.max_stream_data_local:
#print("WRITE STREAM'S LIMITS FRAME") #DEBUG*
buf = builder.start_frame(
QuicFrameType.MAX_STREAM_DATA,
capacity=MAX_STREAM_DATA_FRAME_CAPACITY,
handler=self._on_max_stream_data_delivery,
handler_args=(stream,),
)
buf.push_uint_var(stream.stream_id)
buf.push_uint_var(stream.max_stream_data_local)
stream.max_stream_data_local_sent = stream.max_stream_data_local
# log frame
if self._quic_logger is not None:
builder.quic_logger_frames.append(
self._quic_logger.encode_max_stream_data_frame(
maximum=stream.max_stream_data_local, stream_id=stream.stream_id
)
)
def _write_streams_blocked_frame(
self, builder: QuicPacketBuilder, frame_type: QuicFrameType, limit: int
) -> None:
#print("WRITE STREAMS BLOCKED FRAME") #DEBUG*
buf = builder.start_frame(frame_type, capacity=STREAMS_BLOCKED_CAPACITY)
buf.push_uint_var(limit)
# log frame
if self._quic_logger is not None:
builder.quic_logger_frames.append(
self._quic_logger.encode_streams_blocked_frame(
is_unidirectional=frame_type == QuicFrameType.STREAMS_BLOCKED_UNI,
limit=limit,
)
)