(self, pkt, seq_num=None, iv=None, esn_en=None, esn=None)
| 1017 | (pkt.spi, self.spi)) |
| 1018 | |
| 1019 | def _encrypt_esp(self, pkt, seq_num=None, iv=None, esn_en=None, esn=None): |
| 1020 | |
| 1021 | if iv is None: |
| 1022 | iv = self.crypt_algo.generate_iv() |
| 1023 | else: |
| 1024 | if len(iv) != self.crypt_algo.iv_size: |
| 1025 | raise TypeError('iv length must be %s' % self.crypt_algo.iv_size) # noqa: E501 |
| 1026 | |
| 1027 | esp = _ESPPlain(spi=self.spi, seq=seq_num or self.seq_num, iv=iv) |
| 1028 | |
| 1029 | if self.tunnel_header: |
| 1030 | tunnel = self.tunnel_header.copy() |
| 1031 | |
| 1032 | if tunnel.version == 4: |
| 1033 | del tunnel.proto |
| 1034 | del tunnel.len |
| 1035 | del tunnel.chksum |
| 1036 | else: |
| 1037 | del tunnel.nh |
| 1038 | del tunnel.plen |
| 1039 | |
| 1040 | pkt = tunnel.__class__(raw(tunnel / pkt)) |
| 1041 | |
| 1042 | ip_header, nh, payload = split_for_transport(pkt, socket.IPPROTO_ESP) |
| 1043 | esp.data = payload |
| 1044 | esp.nh = nh |
| 1045 | |
| 1046 | esp = self.crypt_algo.pad(esp) |
| 1047 | esp = self.crypt_algo.encrypt(self, esp, self.crypt_key, |
| 1048 | self.crypt_icv_size, |
| 1049 | esn_en=esn_en or self.esn_en, |
| 1050 | esn=esn or self.esn) |
| 1051 | |
| 1052 | self.auth_algo.sign(esp, |
| 1053 | self.auth_key, |
| 1054 | esn_en=esn_en or self.esn_en, |
| 1055 | esn=esn or self.esn) |
| 1056 | |
| 1057 | if self.nat_t_header: |
| 1058 | nat_t_header = self.nat_t_header.copy() |
| 1059 | nat_t_header.chksum = 0 |
| 1060 | del nat_t_header.len |
| 1061 | if ip_header.version == 4: |
| 1062 | del ip_header.proto |
| 1063 | else: |
| 1064 | del ip_header.nh |
| 1065 | ip_header /= nat_t_header |
| 1066 | |
| 1067 | if ip_header.version == 4: |
| 1068 | del ip_header.len |
| 1069 | del ip_header.chksum |
| 1070 | else: |
| 1071 | del ip_header.plen |
| 1072 | |
| 1073 | # sequence number must always change, unless specified by the user |
| 1074 | if seq_num is None: |
| 1075 | self.seq_num += 1 |
| 1076 |
no test coverage detected