(
&self,
args: GetCertArgs,
vm: &VirtualMachine,
)
| 3742 | |
| 3743 | #[pymethod] |
| 3744 | fn getpeercert( |
| 3745 | &self, |
| 3746 | args: GetCertArgs, |
| 3747 | vm: &VirtualMachine, |
| 3748 | ) -> PyResult<Option<PyObjectRef>> { |
| 3749 | let binary = args.binary_form.unwrap_or(false); |
| 3750 | |
| 3751 | // Check if handshake is complete |
| 3752 | if !*self.handshake_done.lock() { |
| 3753 | return Err(vm.new_value_error("handshake not done yet")); |
| 3754 | } |
| 3755 | |
| 3756 | // Extract DER bytes from connection, releasing lock quickly |
| 3757 | let der_bytes = { |
| 3758 | let conn_guard = self.connection.lock(); |
| 3759 | let conn = conn_guard |
| 3760 | .as_ref() |
| 3761 | .ok_or_else(|| vm.new_value_error("No TLS connection established"))?; |
| 3762 | |
| 3763 | let Some(peer_certificates) = conn.peer_certificates() else { |
| 3764 | return Ok(None); |
| 3765 | }; |
| 3766 | let cert = peer_certificates |
| 3767 | .first() |
| 3768 | .ok_or_else(|| vm.new_value_error("No peer certificate available"))?; |
| 3769 | cert.as_ref().to_vec() |
| 3770 | }; |
| 3771 | |
| 3772 | if binary { |
| 3773 | // Return DER-encoded certificate as bytes |
| 3774 | return Ok(Some(vm.ctx.new_bytes(der_bytes).into())); |
| 3775 | } |
| 3776 | |
| 3777 | // Dictionary mode: check verify_mode |
| 3778 | let verify_mode = *self.context.read().verify_mode.read(); |
| 3779 | |
| 3780 | if verify_mode == CERT_NONE { |
| 3781 | // Return empty dict when CERT_NONE |
| 3782 | return Ok(Some(vm.ctx.new_dict().into())); |
| 3783 | } |
| 3784 | |
| 3785 | // Parse DER certificate and convert to dict (outside lock) |
| 3786 | let (_, cert) = x509_parser::parse_x509_certificate(&der_bytes) |
| 3787 | .map_err(|e| vm.new_value_error(format!("Failed to parse certificate: {e}")))?; |
| 3788 | |
| 3789 | cert::cert_to_dict(vm, &cert).map(Some) |
| 3790 | } |
| 3791 | |
| 3792 | #[pymethod] |
| 3793 | fn cipher(&self) -> Option<(String, String, i32)> { |
nothing calls this directly
no test coverage detected