// if openssl is vendored, it doesn't know the locations of system certificates

#[cfg(feature = "ssl-vendor")]

if let None | Some("0") = option_env!("OPENSSL_NO_VENDOR") {

openssl_probe::init_ssl_cert_env_vars();

}

openssl::init();

let module = _ssl::make_module(vm);

#[cfg(ossl101)]

ossl101::extend_module(vm, &module);

#[cfg(ossl111)]

ossl101::extend_module(vm, &module);

#[cfg(windows)]

windows::extend_module(vm, &module);

module

use super::bio;

use crate::{

common::{

ascii,

lock::{PyRwLock, PyRwLockWriteGuard},

},

socket::{self, PySocket},

vm::{

builtins::{PyBaseExceptionRef, PyStrRef, PyType, PyTypeRef, PyWeak},

convert::{ToPyException, ToPyObject},

exceptions,

function::{

ArgBytesLike, ArgCallable, ArgMemoryBuffer, ArgStrOrBytesLike, Either, OptionalArg,

},

stdlib::os::PyPathLike,

types::Constructor,

utils::ToCString,

PyObjectRef, PyPayload, PyRef, PyResult, VirtualMachine,

},

};

use crossbeam_utils::atomic::AtomicCell;

use foreign_types_shared::{ForeignType, ForeignTypeRef};

use openssl::{

asn1::{Asn1Object, Asn1ObjectRef},

error::ErrorStack,

nid::Nid,

ssl::{self, SslContextBuilder, SslOptions, SslVerifyMode},

x509::{self, X509Ref, X509},

};

use openssl_sys as sys;

use std::{

ffi::CStr,

fmt,

io::{Read, Write},

time::Instant,

};

// Constants

#[pyattr]

use sys::{

SSL_OP_NO_SSLv2 as OP_NO_SSLv2,

SSL_OP_NO_SSLv3 as OP_NO_SSLv3,

SSL_OP_NO_TLSv1 as OP_NO_TLSv1,

// TODO: so many more of these

SSL_AD_DECODE_ERROR as ALERT_DESCRIPTION_DECODE_ERROR,

SSL_AD_ILLEGAL_PARAMETER as ALERT_DESCRIPTION_ILLEGAL_PARAMETER,

SSL_AD_UNRECOGNIZED_NAME as ALERT_DESCRIPTION_UNRECOGNIZED_NAME,

// SSL_ERROR_INVALID_ERROR_CODE,

SSL_ERROR_SSL,

// SSL_ERROR_WANT_X509_LOOKUP,

SSL_ERROR_SYSCALL,

SSL_ERROR_WANT_CONNECT,

SSL_ERROR_WANT_READ,

SSL_ERROR_WANT_WRITE,

// #ifdef SSL_OP_SINGLE_ECDH_USE

// SSL_OP_SINGLE_ECDH_USE as OP_SINGLE_ECDH_USE

// #endif

// X509_V_FLAG_CRL_CHECK as VERIFY_CRL_CHECK_LEAF,

// sys::X509_V_FLAG_CRL_CHECK|sys::X509_V_FLAG_CRL_CHECK_ALL as VERIFY_CRL_CHECK_CHAIN

// X509_V_FLAG_X509_STRICT as VERIFY_X509_STRICT,

SSL_ERROR_ZERO_RETURN,

SSL_OP_CIPHER_SERVER_PREFERENCE as OP_CIPHER_SERVER_PREFERENCE,

SSL_OP_NO_TICKET as OP_NO_TICKET,

SSL_OP_SINGLE_DH_USE as OP_SINGLE_DH_USE,

};

// taken from CPython, should probably be kept up to date with their version if it ever changes

#[pyattr]

const _DEFAULT_CIPHERS: &str =

"DEFAULT:!aNULL:!eNULL:!MD5:!3DES:!DES:!RC4:!IDEA:!SEED:!aDSS:!SRP:!PSK";

// #[pyattr] PROTOCOL_SSLv2: u32 = SslVersion::Ssl2 as u32; // unsupported

// #[pyattr] PROTOCOL_SSLv3: u32 = SslVersion::Ssl3 as u32;

#[pyattr]

const PROTOCOL_SSLv23: u32 = SslVersion::Tls as u32;

#[pyattr]

const PROTOCOL_TLS: u32 = SslVersion::Tls as u32;

#[pyattr]

const PROTOCOL_TLS_CLIENT: u32 = SslVersion::TlsClient as u32;

#[pyattr]

const PROTOCOL_TLS_SERVER: u32 = SslVersion::TlsServer as u32;

#[pyattr]

const PROTOCOL_TLSv1: u32 = SslVersion::Tls1 as u32;

#[pyattr]

const PROTO_MINIMUM_SUPPORTED: i32 = ProtoVersion::MinSupported as i32;

#[pyattr]

const PROTO_SSLv3: i32 = ProtoVersion::Ssl3 as i32;

#[pyattr]

const PROTO_TLSv1: i32 = ProtoVersion::Tls1 as i32;

#[pyattr]

const PROTO_TLSv1_1: i32 = ProtoVersion::Tls1_1 as i32;

#[pyattr]

const PROTO_TLSv1_2: i32 = ProtoVersion::Tls1_2 as i32;

#[pyattr]

const PROTO_TLSv1_3: i32 = ProtoVersion::Tls1_3 as i32;

#[pyattr]

const PROTO_MAXIMUM_SUPPORTED: i32 = ProtoVersion::MaxSupported as i32;

#[pyattr]

const OP_ALL: libc::c_ulong = sys::SSL_OP_ALL & !sys::SSL_OP_DONT_INSERT_EMPTY_FRAGMENTS;

#[pyattr]

const HAS_TLS_UNIQUE: bool = true;

#[pyattr]

const CERT_NONE: u32 = CertRequirements::None as u32;

#[pyattr]

const CERT_OPTIONAL: u32 = CertRequirements::Optional as u32;

#[pyattr]

const CERT_REQUIRED: u32 = CertRequirements::Required as u32;

#[pyattr]

const VERIFY_DEFAULT: u32 = 0;

#[pyattr]

const SSL_ERROR_EOF: u32 = 8; // custom for python

#[pyattr]

const HAS_SNI: bool = true;

#[pyattr]

const HAS_ECDH: bool = false;

#[pyattr]

const HAS_NPN: bool = false;

#[pyattr]

const HAS_ALPN: bool = true;

#[pyattr]

const HAS_SSLv2: bool = true;

#[pyattr]

const HAS_SSLv3: bool = true;

#[pyattr]

const HAS_TLSv1: bool = true;

#[pyattr]

const HAS_TLSv1_1: bool = true;

#[pyattr]

const HAS_TLSv1_2: bool = true;

#[pyattr]

const HAS_TLSv1_3: bool = cfg!(ossl111);

// the openssl version from the API headers

#[pyattr(name = "OPENSSL_VERSION")]

fn openssl_version(_vm: &VirtualMachine) -> &str {

openssl::version::version()

}

#[pyattr(name = "OPENSSL_VERSION_NUMBER")]

fn openssl_version_number(_vm: &VirtualMachine) -> i64 {

openssl::version::number()

}

#[pyattr(name = "OPENSSL_VERSION_INFO")]

fn openssl_version_info(_vm: &VirtualMachine) -> OpensslVersionInfo {

parse_version_info(openssl::version::number())

}

#[pyattr(name = "_OPENSSL_API_VERSION")]

fn _openssl_api_version(_vm: &VirtualMachine) -> OpensslVersionInfo {

let openssl_api_version = i64::from_str_radix(env!("OPENSSL_API_VERSION"), 16).unwrap();

parse_version_info(openssl_api_version)

}

/// An error occurred in the SSL implementation.

#[pyattr(name = "SSLError", once)]

fn ssl_error(vm: &VirtualMachine) -> PyTypeRef {

vm.ctx.new_exception_type(

"ssl",

"SSLError",

Some(vec![vm.ctx.exceptions.os_error.to_owned()]),

)

}

/// A certificate could not be verified.

#[pyattr(name = "SSLCertVerificationError", once)]

fn ssl_cert_verification_error(vm: &VirtualMachine) -> PyTypeRef {

vm.ctx.new_exception_type(

"ssl",

"SSLCertVerificationError",

Some(vec![

ssl_error(vm),

vm.ctx.exceptions.value_error.to_owned(),

]),

)

}

/// SSL/TLS session closed cleanly.

#[pyattr(name = "SSLZeroReturnError", once)]

fn ssl_zero_return_error(vm: &VirtualMachine) -> PyTypeRef {

vm.ctx

.new_exception_type("ssl", "SSLZeroReturnError", Some(vec![ssl_error(vm)]))

}

/// Non-blocking SSL socket needs to read more data before the requested operation can be completed.

#[pyattr(name = "SSLWantReadError", once)]

fn ssl_want_read_error(vm: &VirtualMachine) -> PyTypeRef {

vm.ctx

.new_exception_type("ssl", "SSLWantReadError", Some(vec![ssl_error(vm)]))

}

/// Non-blocking SSL socket needs to write more data before the requested operation can be completed.

#[pyattr(name = "SSLWantWriteError", once)]

fn ssl_want_write_error(vm: &VirtualMachine) -> PyTypeRef {

vm.ctx

.new_exception_type("ssl", "SSLWantWriteError", Some(vec![ssl_error(vm)]))

}

/// System error when attempting SSL operation.

#[pyattr(name = "SSLSyscallError", once)]

fn ssl_syscall_error(vm: &VirtualMachine) -> PyTypeRef {

vm.ctx

.new_exception_type("ssl", "SSLSyscallError", Some(vec![ssl_error(vm)]))

}

/// SSL/TLS connection terminated abruptly.

#[pyattr(name = "SSLEOFError", once)]

fn ssl_eof_error(vm: &VirtualMachine) -> PyTypeRef {

PyType::new_simple_ref("ssl.SSLEOFError", &ssl_error(vm)).unwrap()

}

type OpensslVersionInfo = (u8, u8, u8, u8, u8);

const fn parse_version_info(mut n: i64) -> OpensslVersionInfo {

let status = (n & 0xF) as u8;

n >>= 4;

let patch = (n & 0xFF) as u8;

n >>= 8;

let fix = (n & 0xFF) as u8;

n >>= 8;

let minor = (n & 0xFF) as u8;

n >>= 8;

let major = (n & 0xFF) as u8;

(major, minor, fix, patch, status)

}

#[derive(Copy, Clone, num_enum::IntoPrimitive, num_enum::TryFromPrimitive, PartialEq)]

#[repr(i32)]

enum SslVersion {

Ssl2,

Ssl3 = 1,

Tls,

Tls1,

// TODO: Tls1_1, Tls1_2 ?

TlsClient = 0x10,

TlsServer,

}

#[derive(Copy, Clone, num_enum::IntoPrimitive, num_enum::TryFromPrimitive)]

#[repr(i32)]

enum ProtoVersion {

MinSupported = -2,

Ssl3 = sys::SSL3_VERSION,

Tls1 = sys::TLS1_VERSION,

Tls1_1 = sys::TLS1_1_VERSION,

Tls1_2 = sys::TLS1_2_VERSION,

#[cfg(ossl111)]

Tls1_3 = sys::TLS1_3_VERSION,

#[cfg(not(ossl111))]

Tls1_3 = 0x304,

MaxSupported = -1,

}

#[derive(num_enum::IntoPrimitive, num_enum::TryFromPrimitive)]

#[repr(i32)]

enum CertRequirements {

None,

Optional,

Required,

}

#[derive(Debug, PartialEq)]

enum SslServerOrClient {

Client,

Server,

}

unsafe fn ptr2obj(ptr: *mut sys::ASN1_OBJECT) -> Option<Asn1Object> {

if ptr.is_null() {

None

} else {

Some(Asn1Object::from_ptr(ptr))

}

}

fn _txt2obj(s: &CStr, no_name: bool) -> Option<Asn1Object> {

unsafe { ptr2obj(sys::OBJ_txt2obj(s.as_ptr(), if no_name { 1 } else { 0 })) }

}

fn _nid2obj(nid: Nid) -> Option<Asn1Object> {

unsafe { ptr2obj(sys::OBJ_nid2obj(nid.as_raw())) }

}

fn obj2txt(obj: &Asn1ObjectRef, no_name: bool) -> Option<String> {

let no_name = if no_name { 1 } else { 0 };

let ptr = obj.as_ptr();

let b = unsafe {

let buflen = sys::OBJ_obj2txt(std::ptr::null_mut(), 0, ptr, no_name);

assert!(buflen >= 0);

if buflen == 0 {

return None;

}

let buflen = buflen as usize;

let mut buf = Vec::<u8>::with_capacity(buflen + 1);

let ret = sys::OBJ_obj2txt(

buf.as_mut_ptr() as *mut libc::c_char,

buf.capacity() as _,

ptr,

no_name,

);

assert!(ret >= 0);

// SAFETY: OBJ_obj2txt initialized the buffer successfully

buf.set_len(buflen);

buf

};

let s = String::from_utf8(b)

.unwrap_or_else(|e| String::from_utf8_lossy(e.as_bytes()).into_owned());

Some(s)

}

type PyNid = (libc::c_int, String, String, Option<String>);

fn obj2py(obj: &Asn1ObjectRef) -> PyNid {

let nid = obj.nid();

(

nid.as_raw(),

nid.short_name().unwrap().to_owned(),

nid.long_name().unwrap().to_owned(),

obj2txt(obj, true),

)

}

#[derive(FromArgs)]

struct Txt2ObjArgs {

txt: PyStrRef,

#[pyarg(any, default = "false")]

name: bool,

}

#[pyfunction]

fn txt2obj(args: Txt2ObjArgs, vm: &VirtualMachine) -> PyResult<PyNid> {

_txt2obj(&args.txt.to_cstring(vm)?, !args.name)

.as_deref()

.map(obj2py)

.ok_or_else(|| vm.new_value_error(format!("unknown object '{}'", args.txt)))

}

#[pyfunction]

fn nid2obj(nid: libc::c_int, vm: &VirtualMachine) -> PyResult<PyNid> {

_nid2obj(Nid::from_raw(nid))

.as_deref()

.map(obj2py)

.ok_or_else(|| vm.new_value_error(format!("unknown NID {}", nid)))

}

#[pyfunction]

fn get_default_verify_paths() -> (String, String, String, String) {

macro_rules! convert {

($f:ident) => {

CStr::from_ptr(sys::$f()).to_string_lossy().into_owned()

};

}

unsafe {

(

convert!(X509_get_default_cert_file_env),

convert!(X509_get_default_cert_file),

convert!(X509_get_default_cert_dir_env),

convert!(X509_get_default_cert_dir),

)

}

}

#[pyfunction(name = "RAND_status")]

fn rand_status() -> i32 {

unsafe { sys::RAND_status() }

}

#[pyfunction(name = "RAND_add")]

fn rand_add(string: ArgStrOrBytesLike, entropy: f64) {

let f = |b: &[u8]| {

for buf in b.chunks(libc::c_int::max_value() as usize) {

unsafe { sys::RAND_add(buf.as_ptr() as *const _, buf.len() as _, entropy) }

}

};

f(&string.borrow_bytes())

}

#[pyfunction(name = "RAND_bytes")]

fn rand_bytes(n: i32, vm: &VirtualMachine) -> PyResult<Vec<u8>> {

if n < 0 {

return Err(vm.new_value_error("num must be positive".to_owned()));

}

let mut buf = vec![0; n as usize];

openssl::rand::rand_bytes(&mut buf).map_err(|e| convert_openssl_error(vm, e))?;

Ok(buf)

}

#[pyfunction(name = "RAND_pseudo_bytes")]

fn rand_pseudo_bytes(n: i32, vm: &VirtualMachine) -> PyResult<(Vec<u8>, bool)> {

if n < 0 {

return Err(vm.new_value_error("num must be positive".to_owned()));

}

let mut buf = vec![0; n as usize];

let ret = unsafe { sys::RAND_bytes(buf.as_mut_ptr(), n) };

match ret {

0 | 1 => Ok((buf, ret == 1)),

_ => Err(convert_openssl_error(vm, ErrorStack::get())),

}

}

#[pyattr]

#[pyclass(module = "ssl", name = "_SSLContext")]

#[derive(PyPayload)]

struct PySslContext {

ctx: PyRwLock<SslContextBuilder>,

check_hostname: AtomicCell<bool>,

protocol: SslVersion,

}

impl fmt::Debug for PySslContext {

fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {

f.pad("_SSLContext")

}

}

fn builder_as_ctx(x: &SslContextBuilder) -> &ssl::SslContextRef {

unsafe { ssl::SslContextRef::from_ptr(x.as_ptr()) }

}

impl Constructor for PySslContext {

type Args = i32;

fn py_new(cls: PyTypeRef, proto_version: Self::Args, vm: &VirtualMachine) -> PyResult {

let proto = SslVersion::try_from(proto_version)

.map_err(|_| vm.new_value_error("invalid protocol version".to_owned()))?;

let method = match proto {

// SslVersion::Ssl3 => unsafe { ssl::SslMethod::from_ptr(sys::SSLv3_method()) },

SslVersion::Tls => ssl::SslMethod::tls(),

// TODO: Tls1_1, Tls1_2 ?

SslVersion::TlsClient => ssl::SslMethod::tls_client(),

SslVersion::TlsServer => ssl::SslMethod::tls_server(),

_ => return Err(vm.new_value_error("invalid protocol version".to_owned())),

};

let mut builder =

SslContextBuilder::new(method).map_err(|e| convert_openssl_error(vm, e))?;

#[cfg(target_os = "android")]

android::load_client_ca_list(vm, &mut builder)?;

let check_hostname = proto == SslVersion::TlsClient;

builder.set_verify(if check_hostname {

SslVerifyMode::PEER | SslVerifyMode::FAIL_IF_NO_PEER_CERT

} else {

SslVerifyMode::NONE

});

let mut options = SslOptions::ALL & !SslOptions::DONT_INSERT_EMPTY_FRAGMENTS;

if proto != SslVersion::Ssl2 {

options |= SslOptions::NO_SSLV2;

}

if proto != SslVersion::Ssl3 {

options |= SslOptions::NO_SSLV3;

}

options |= SslOptions::NO_COMPRESSION;

options |= SslOptions::CIPHER_SERVER_PREFERENCE;

options |= SslOptions::SINGLE_DH_USE;

options |= SslOptions::SINGLE_ECDH_USE;

builder.set_options(options);

let mode = ssl::SslMode::ACCEPT_MOVING_WRITE_BUFFER | ssl::SslMode::AUTO_RETRY;

builder.set_mode(mode);

#[cfg(ossl111)]

unsafe {

sys::SSL_CTX_set_post_handshake_auth(builder.as_ptr(), 0);

}

builder

.set_session_id_context(b"Python")

.map_err(|e| convert_openssl_error(vm, e))?;

PySslContext {

ctx: PyRwLock::new(builder),

check_hostname: AtomicCell::new(check_hostname),

protocol: proto,

}

.into_ref_with_type(vm, cls)

.map(Into::into)

}

}

#[pyimpl(flags(BASETYPE), with(Constructor))]

impl PySslContext {

fn builder(&self) -> PyRwLockWriteGuard<'_, SslContextBuilder> {

self.ctx.write()

}

fn exec_ctx<F, R>(&self, func: F) -> R

where

F: Fn(&ssl::SslContextRef) -> R,

{

let c = self.ctx.read();

func(builder_as_ctx(&c))

}

#[pymethod]

fn set_ciphers(&self, cipherlist: PyStrRef, vm: &VirtualMachine) -> PyResult<()> {

let ciphers = cipherlist.as_str();

if ciphers.contains('\0') {

return Err(exceptions::cstring_error(vm));

}

self.builder().set_cipher_list(ciphers).map_err(|_| {

vm.new_exception_msg(ssl_error(vm), "No cipher can be selected.".to_owned())

})

}

#[pyproperty]

fn options(&self) -> libc::c_ulong {

self.ctx.read().options().bits()

}

#[pyproperty(setter)]

fn set_options(&self, opts: libc::c_ulong) {

self.builder()

.set_options(SslOptions::from_bits_truncate(opts));

}

#[pyproperty]

fn protocol(&self) -> i32 {

self.protocol as i32

}

#[pyproperty]

fn verify_mode(&self) -> i32 {

let mode = self.exec_ctx(|ctx| ctx.verify_mode());

if mode == SslVerifyMode::NONE {

CertRequirements::None.into()

} else if mode == SslVerifyMode::PEER {

CertRequirements::Optional.into()

} else if mode == SslVerifyMode::PEER | SslVerifyMode::FAIL_IF_NO_PEER_CERT {

CertRequirements::Required.into()

} else {

unreachable!()

}

}

#[pyproperty(setter)]

fn set_verify_mode(&self, cert: i32, vm: &VirtualMachine) -> PyResult<()> {

let mut ctx = self.builder();

let cert_req = CertRequirements::try_from(cert)

.map_err(|_| vm.new_value_error("invalid value for verify_mode".to_owned()))?;

let mode = match cert_req {

CertRequirements::None if self.check_hostname.load() => {

return Err(vm.new_value_error(

"Cannot set verify_mode to CERT_NONE when check_hostname is enabled."

.to_owned(),

))

}

CertRequirements::None => SslVerifyMode::NONE,

CertRequirements::Optional => SslVerifyMode::PEER,

CertRequirements::Required => {

SslVerifyMode::PEER | SslVerifyMode::FAIL_IF_NO_PEER_CERT

}

};

ctx.set_verify(mode);

Ok(())

}

#[pyproperty]

fn check_hostname(&self) -> bool {

self.check_hostname.load()

}

#[pyproperty(setter)]

fn set_check_hostname(&self, ch: bool) {

let mut ctx = self.builder();

if ch && builder_as_ctx(&ctx).verify_mode() == SslVerifyMode::NONE {

ctx.set_verify(SslVerifyMode::PEER | SslVerifyMode::FAIL_IF_NO_PEER_CERT);

}

self.check_hostname.store(ch);

}

#[pymethod]

fn set_default_verify_paths(&self, vm: &VirtualMachine) -> PyResult<()> {

self.builder()

.set_default_verify_paths()

.map_err(|e| convert_openssl_error(vm, e))

}

#[pymethod]

fn _set_alpn_protocols(&self, protos: ArgBytesLike, vm: &VirtualMachine) -> PyResult<()> {

#[cfg(ossl102)]

{

let mut ctx = self.builder();

let server = protos.with_ref(|pbuf| {

if pbuf.len() > libc::c_uint::MAX as usize {

return Err(vm.new_overflow_error(format!(

"protocols longer than {} bytes",

libc::c_uint::MAX

)));

}

ctx.set_alpn_protos(pbuf)

.map_err(|e| convert_openssl_error(vm, e))?;

Ok(pbuf.to_vec())

})?;

ctx.set_alpn_select_callback(move |_, client| {

ssl::select_next_proto(&server, client).ok_or(ssl::AlpnError::NOACK)

});

Ok(())

}

#[cfg(not(ossl102))]

{

Err(vm.new_not_implemented_error(

"The NPN extension requires OpenSSL 1.0.1 or later.".to_owned(),

))

}

}

#[pymethod]

fn load_verify_locations(

&self,

args: LoadVerifyLocationsArgs,

vm: &VirtualMachine,

) -> PyResult<()> {

if let (None, None, None) = (&args.cafile, &args.capath, &args.cadata) {

return Err(

vm.new_type_error("cafile, capath and cadata cannot be all omitted".to_owned())

);

}

if let Some(cadata) = args.cadata {

let certs = match cadata {

Either::A(s) => {

if !s.is_ascii() {

return Err(vm.new_type_error("Must be an ascii string".to_owned()));

}

X509::stack_from_pem(s.as_str().as_bytes())

}

Either::B(b) => b.with_ref(x509_stack_from_der),

};

let certs = certs.map_err(|e| convert_openssl_error(vm, e))?;

let mut ctx = self.builder();

let store = ctx.cert_store_mut();

for cert in certs {

store

.add_cert(cert)

.map_err(|e| convert_openssl_error(vm, e))?;

}

}

if args.cafile.is_some() || args.capath.is_some() {

let cafile = args.cafile.map(|s| s.to_cstring(vm)).transpose()?;

let capath = args.capath.map(|s| s.to_cstring(vm)).transpose()?;

let ret = unsafe {

let ctx = self.ctx.write();

sys::SSL_CTX_load_verify_locations(

ctx.as_ptr(),

cafile

.as_ref()

.map_or_else(std::ptr::null, |cs| cs.as_ptr()),

capath

.as_ref()

.map_or_else(std::ptr::null, |cs| cs.as_ptr()),

)

};

if ret != 1 {

let errno = std::io::Error::last_os_error().raw_os_error().unwrap();

let err = if errno != 0 {

crate::vm::stdlib::os::errno_err(vm)

} else {

convert_openssl_error(vm, ErrorStack::get())

};

return Err(err);

}

}

Ok(())

}

#[pymethod]

fn get_ca_certs(

&self,

binary_form: OptionalArg<bool>,

vm: &VirtualMachine,

) -> PyResult<Vec<PyObjectRef>> {

let binary_form = binary_form.unwrap_or(false);

self.exec_ctx(|ctx| {

let certs = ctx

.cert_store()

.objects()

.iter()

.filter_map(|obj| obj.x509())

.map(|cert| cert_to_py(vm, cert, binary_form))

.collect::<Result<Vec<_>, _>>()?;

Ok(certs)

})

}

#[pymethod]

fn load_cert_chain(&self, args: LoadCertChainArgs, vm: &VirtualMachine) -> PyResult<()> {

let LoadCertChainArgs {

certfile,

keyfile,

password,

} = args;

// TODO: requires passing a callback to C

if password.is_some() {

return Err(

vm.new_not_implemented_error("password arg not yet supported".to_owned())

);

}

let mut ctx = self.builder();

ctx.set_certificate_chain_file(&certfile)

.and_then(|()| {

ctx.set_private_key_file(

keyfile.as_ref().unwrap_or(&certfile),

ssl::SslFiletype::PEM,

)

})

.and_then(|()| ctx.check_private_key())

.map_err(|e| convert_openssl_error(vm, e))

}

#[pymethod]

fn _wrap_socket(

zelf: PyRef<Self>,

args: WrapSocketArgs,

vm: &VirtualMachine,

) -> PyResult<PySslSocket> {

let mut ssl = zelf

.exec_ctx(ssl::Ssl::new)

.map_err(|e| convert_openssl_error(vm, e))?;

let socket_type = if args.server_side {

ssl.set_accept_state();

SslServerOrClient::Server

} else {

ssl.set_connect_state();

SslServerOrClient::Client

};

if let Some(hostname) = &args.server_hostname {

let hostname = hostname.as_str();

if hostname.is_empty() || hostname.starts_with('.') {

return Err(vm.new_value_error(

"server_hostname cannot be an empty string or start with a leading dot."

.to_owned(),

));

}

let ip = hostname.parse::<std::net::IpAddr>();

if ip.is_err() {

ssl.set_hostname(hostname)

.map_err(|e| convert_openssl_error(vm, e))?;

}

if zelf.check_hostname.load() {

if let Ok(ip) = ip {

ssl.param_mut()

.set_ip(ip)

.map_err(|e| convert_openssl_error(vm, e))?;

} else {

ssl.param_mut()

.set_host(hostname)

.map_err(|e| convert_openssl_error(vm, e))?;

}

}

}

let stream = ssl::SslStream::new(ssl, SocketStream(args.sock.clone()))

.map_err(|e| convert_openssl_error(vm, e))?;

// TODO: use this

let _ = args.session;

Ok(PySslSocket {

ctx: zelf,

stream: PyRwLock::new(stream),

socket_type,

server_hostname: args.server_hostname,

owner: PyRwLock::new(args.owner.map(|o| o.downgrade(None, vm)).transpose()?),

})

}

}

#[derive(FromArgs)]

struct WrapSocketArgs {

sock: PyRef<PySocket>,

server_side: bool,

#[pyarg(any, default)]

server_hostname: Option<PyStrRef>,

#[pyarg(named, default)]

owner: Option<PyObjectRef>,

#[pyarg(named, default)]

session: Option<PyObjectRef>,

}

#[derive(FromArgs)]

struct LoadVerifyLocationsArgs {

#[pyarg(any, default)]

cafile: Option<PyStrRef>,

#[pyarg(any, default)]

capath: Option<PyStrRef>,

#[pyarg(any, default)]

cadata: Option<Either<PyStrRef, ArgBytesLike>>,

}