Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/pylibsrtp/__init__.py: 61%

1from socket import htonl 

2from typing import Optional 

3 

4from ._binding import ffi, lib 

5 

6__all__ = ["Error", "Policy", "Session", "__version__"] 

7__version__ = "1.0.0" 

8 

9 

10ERRORS = [ 

11 "nothing to report", 

12 "unspecified failure", 

13 "unsupported parameter", 

14 "couldn't allocate memory", 

15 "couldn't deallocate properly", 

16 "couldn't initialize", 

17 "can't process as much data as requested", 

18 "authentication failure", 

19 "cipher failure", 

20 "replay check failed (bad index)", 

21 "replay check failed (index too old)", 

22 "algorithm failed test routine", 

23 "unsupported operation", 

24 "no appropriate context found", 

25 "unable to perform desired validation", 

26 "can't use key any more", 

27 "error in use of socket", 

28 "error in use POSIX signals", 

29 "nonce check failed", 

30 "couldn't read data", 

31 "couldn't write data", 

32 "error parsing data", 

33 "error encoding data", 

34 "error while using semaphores", 

35 "error while using pfkey", 

36 "error MKI present in packet is invalid", 

37 "packet index is too old to consider", 

38 "packet index advanced, reset needed", 

39] 

40 

41 

42# SRTP_MAX_TAG_LEN + SRTP_MAX_MKI_LEN 

43SRTP_MAX_TRAILER_LEN = 16 + 128 

44 

45# SRTP_SRCTP_INDEX_LEN + SRTP_MAX_TAG_LEN + SRTP_MAX_MKI_LEN 

46SRTP_MAX_SRTCP_TRAILER_LEN = 4 + 16 + 128 

47 

48 

49class Error(Exception): 

50 """ 

51 Error that occurred making a `libsrtp` API call. 

52 """ 

53 

54 pass 

55 

56 

57def _srtp_assert(rc): 

58 if rc != lib.srtp_err_status_ok: 

59 raise Error(ERRORS[rc]) 

60 

61 

62class Policy: 

63 """ 

64 Policy for single SRTP stream. 

65 """ 

66 

67 #: AES-128 CM mode with 80-bit authentication tag (default) 

68 SRTP_PROFILE_AES128_CM_SHA1_80 = lib.srtp_profile_aes128_cm_sha1_80 

69 #: AES-128 CM mode with 32-bit authentication tag 

70 SRTP_PROFILE_AES128_CM_SHA1_32 = lib.srtp_profile_aes128_cm_sha1_32 

71 #: AES-128 GCM mode 

72 SRTP_PROFILE_AEAD_AES_128_GCM = lib.srtp_profile_aead_aes_128_gcm 

73 #: AES-256 GCM mode 

74 SRTP_PROFILE_AEAD_AES_256_GCM = lib.srtp_profile_aead_aes_256_gcm 

75 

76 #: Indicates an undefined SSRC type 

77 SSRC_UNDEFINED = lib.ssrc_undefined 

78 #: Indicates a specific SSRC value 

79 SSRC_SPECIFIC = lib.ssrc_specific 

80 #: Indicates any inbound SSRC value 

81 SSRC_ANY_INBOUND = lib.ssrc_any_inbound 

82 #: Indicates any inbound SSRC value 

83 SSRC_ANY_OUTBOUND = lib.ssrc_any_outbound 

84 

85 def __init__( 

86 self, 

87 key: Optional[bytes] = None, 

88 ssrc_type: int = SSRC_UNDEFINED, 

89 ssrc_value: int = 0, 

90 srtp_profile: int = SRTP_PROFILE_AES128_CM_SHA1_80, 

91 ) -> None: 

92 self._policy = ffi.new("srtp_policy_t *") 

93 self._srtp_profile = srtp_profile 

94 

95 _srtp_assert( 

96 lib.srtp_crypto_policy_set_from_profile_for_rtp( 

97 ffi.addressof(self._policy.rtp), srtp_profile 

98 ) 

99 ) 

100 _srtp_assert( 

101 lib.srtp_crypto_policy_set_from_profile_for_rtcp( 

102 ffi.addressof(self._policy.rtcp), srtp_profile 

103 ) 

104 ) 

105 

106 self.key = key 

107 self.ssrc_type = ssrc_type 

108 self.ssrc_value = ssrc_value 

109 

110 @property 

111 def allow_repeat_tx(self) -> bool: 

112 """ 

113 Whether retransmissions of packets with the same sequence number are allowed. 

114 """ 

115 return self._policy.allow_repeat_tx == 1 

116 

117 @allow_repeat_tx.setter 

118 def allow_repeat_tx(self, allow_repeat_tx: bool) -> None: 

119 self._policy.allow_repeat_tx = 1 if allow_repeat_tx else 0 

120 

121 @property 

122 def key(self) -> Optional[bytes]: 

123 """ 

124 The SRTP master key + master salt. 

125 """ 

126 if self.__cdata is None: 

127 return None 

128 return ffi.buffer(self.__cdata) 

129 

130 @key.setter 

131 def key(self, key: Optional[bytes]) -> None: 

132 if key is None: 

133 # Clear the key. 

134 self.__cdata = None 

135 self._policy.key = ffi.NULL 

136 return 

137 

138 # Check the key is acceptable then assign it. 

139 expected_length = lib.srtp_profile_get_master_key_length( 

140 self._srtp_profile 

141 ) + lib.srtp_profile_get_master_salt_length(self._srtp_profile) 

142 if not isinstance(key, bytes): 

143 raise TypeError("key must be bytes") 

144 if len(key) < expected_length: 

145 raise ValueError("key must contain at least %d bytes" % expected_length) 

146 self.__cdata = ffi.new("unsigned char[]", len(key)) 

147 self.__cdata[0 : len(key)] = key 

148 self._policy.key = self.__cdata 

149 

150 @property 

151 def srtp_profile(self) -> int: 

152 """ 

153 The SRTP profile. 

154 """ 

155 return self._srtp_profile 

156 

157 @property 

158 def ssrc_type(self) -> int: 

159 """ 

160 The SSRC type. 

161 """ 

162 return self._policy.ssrc.type 

163 

164 @ssrc_type.setter 

165 def ssrc_type(self, ssrc_type: int) -> None: 

166 self._policy.ssrc.type = ssrc_type 

167 

168 @property 

169 def ssrc_value(self) -> int: 

170 """ 

171 The SSRC value, if it is not a wildcard. 

172 """ 

173 return self._policy.ssrc.value 

174 

175 @ssrc_value.setter 

176 def ssrc_value(self, ssrc_value: int) -> None: 

177 self._policy.ssrc.value = ssrc_value 

178 

179 @property 

180 def window_size(self) -> int: 

181 """ 

182 The window size to use for replay protection. 

183 """ 

184 return self._policy.window_size 

185 

186 @window_size.setter 

187 def window_size(self, window_size: int) -> None: 

188 self._policy.window_size = window_size 

189 

190 

191class Session: 

192 """ 

193 SRTP session, which may comprise several streams. 

194 

195 If `policy` is not specified, streams should be added later using the 

196 :func:`add_stream` method. 

197 """ 

198 

199 def __init__(self, policy: Optional[Policy] = None) -> None: 

200 srtp = ffi.new("srtp_t *") 

201 

202 if policy is None: 

203 _policy = ffi.NULL 

204 else: 

205 _policy = policy._policy 

206 _srtp_assert(lib.srtp_create(srtp, _policy)) 

207 

208 self._cdata = ffi.new("char[]", 1500) 

209 self._buffer = ffi.buffer(self._cdata) 

210 self._srtp = ffi.gc(srtp, lambda x: lib.srtp_dealloc(x[0])) 

211 

212 def add_stream(self, policy: Policy) -> None: 

213 """ 

214 Add a stream to the SRTP session, applying the given `policy` 

215 to the stream. 

216 

217 :param policy: :class:`Policy` 

218 """ 

219 _srtp_assert(lib.srtp_add_stream(self._srtp[0], policy._policy)) 

220 

221 def remove_stream(self, ssrc: int) -> None: 

222 """ 

223 Remove the stream with the given `ssrc` from the SRTP session. 

224 

225 :param ssrc: :class:`int` 

226 """ 

227 _srtp_assert(lib.srtp_remove_stream(self._srtp[0], htonl(ssrc))) 

228 

229 def protect(self, packet: bytes) -> bytes: 

230 """ 

231 Apply SRTP protection to the RTP `packet`. 

232 

233 :param packet: :class:`bytes` 

234 :rtype: :class:`bytes` 

235 """ 

236 return self.__process(packet, lib.srtp_protect, SRTP_MAX_TRAILER_LEN) 

237 

238 def protect_rtcp(self, packet: bytes) -> bytes: 

239 """ 

240 Apply SRTCP protection to the RTCP `packet`. 

241 

242 :param packet: :class:`bytes` 

243 :rtype: :class:`bytes` 

244 """ 

245 return self.__process(packet, lib.srtp_protect_rtcp, SRTP_MAX_SRTCP_TRAILER_LEN) 

246 

247 def unprotect(self, packet: bytes) -> bytes: 

248 """ 

249 Verify SRTP protection of the SRTP packet. 

250 

251 :param packet: :class:`bytes` 

252 :rtype: :class:`bytes` 

253 """ 

254 return self.__process(packet, lib.srtp_unprotect) 

255 

256 def unprotect_rtcp(self, packet: bytes) -> bytes: 

257 """ 

258 Verify SRTCP protection of the SRTCP packet. 

259 

260 :param packet: :class:`bytes` 

261 :rtype: :class:`bytes` 

262 """ 

263 return self.__process(packet, lib.srtp_unprotect_rtcp) 

264 

265 def __process(self, data, func, trailer=0): 

266 if not isinstance(data, bytes): 

267 raise TypeError("packet must be bytes") 

268 if len(data) > len(self._cdata) - trailer: 

269 raise ValueError("packet is too long") 

270 

271 len_p = ffi.new("int *") 

272 len_p[0] = len(data) 

273 self._buffer[0 : len(data)] = data 

274 _srtp_assert(func(self._srtp[0], self._cdata, len_p)) 

275 return self._buffer[0 : len_p[0]] 

276 

277 

278lib.srtp_init()