Coverage for /pythoncovmergedfiles/medio/medio/src/selkies/tests/fuzz/fuzz_sctp.py: 58%

1###### Coverage stub 

2import atexit 

3import coverage 

4cov = coverage.coverage(data_file='.coverage', cover_pylib=True) 

5cov.start() 

6# Register an exist handler that will print coverage 

7def exit_handler(): 

8 cov.stop() 

9 cov.save() 

10atexit.register(exit_handler) 

11####### End of coverage stub 

12#!/usr/bin/env python3 

13# This Source Code Form is subject to the terms of the Mozilla Public 

14# License, v. 2.0. If a copy of the MPL was not distributed with this 

15# file, You can obtain one at https://mozilla.org/MPL/2.0/. 

16"""Fuzzes the SCTP association with the packets a peer sends over the data 

17channel's DTLS transport: parsing, then every chunk handler, on a started 

18server-side association that the first byte's low bit has the harness 

19establish first. The checksum and the verification tag are made to match, so 

20inputs reach the handlers rather than stopping at those checks.""" 

21import asyncio 

22import struct 

23import sys 

24import types 

25 

26import atheris 

27 

28with atheris.instrument_imports(): 

29 from google_crc32c import value as crc32c 

30 from selkies.webrtc.rtcsctptransport import ( 

31 SCTP_STATE_COOKIE, 

32 CookieEchoChunk, 

33 InitChunk, 

34 RTCSctpTransport, 

35 parse_packet, 

36 serialize_packet, 

37 ) 

38 

39 

40class _Dtls: 

41 state = "connected" 

42 transport = types.SimpleNamespace(role="controlled") 

43 

44 def __init__(self): 

45 self.sent = [] 

46 

47 def _register_data_receiver(self, receiver): 

48 pass 

49 

50 def _unregister_data_receiver(self, receiver): 

51 pass 

52 

53 async def _send_data(self, data): 

54 self.sent.append(data) 

55 

56 

57async def establish(sctp, dtls): 

58 init = InitChunk() 

59 init.initiate_tag = 1 

60 init.advertised_rwnd = 131072 

61 init.outbound_streams = init.inbound_streams = 65535 

62 init.initial_tsn = 1000 

63 await sctp._handle_data(serialize_packet(5000, 5000, 0, init)) 

64 cookie = dict(parse_packet(dtls.sent[-1])[3][0].params)[SCTP_STATE_COOKIE] 

65 echo = CookieEchoChunk(body=cookie) 

66 await sctp._handle_data(serialize_packet(5000, 5000, sctp._local_verification_tag, echo)) 

67 

68 

69async def receive(data): 

70 dtls = _Dtls() 

71 sctp = RTCSctpTransport(dtls, port=5000) 

72 await sctp.start(None, 5000) 

73 if data[0] & 1: 

74 await establish(sctp, dtls) 

75 data = data[1:] 

76 # A zero tag is kept for INIT; any other is replaced by the association's. 

77 tag = data[4:8] if data[4:8] == b"\0\0\0\0" else struct.pack("!L", sctp._local_verification_tag) 

78 packet = data[:4] + tag + b"\0\0\0\0" + data[12:] 

79 await sctp._handle_data(packet[:8] + struct.pack("<L", crc32c(packet)) + packet[12:]) 

80 

81 

82def TestOneInput(data): 

83 if len(data) >= 13: 

84 asyncio.run(receive(data)) 

85 

86 

87def main(): 

88 atheris.Setup(sys.argv, TestOneInput) 

89 atheris.Fuzz() 

90 

91 

92if __name__ == "__main__": 

93 main()