RustPython/stdlib/src/scproxy.rs at main · drobit/RustPython · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
pub(crate) use _scproxy::make_module;

#[pymodule]
mod _scproxy {
    // straight-forward port of Modules/_scproxy.c

    use crate::vm::{
        builtins::{PyDictRef, PyStr},
        convert::ToPyObject,
        PyResult, VirtualMachine,
    };
    use system_configuration::core_foundation::{
        array::CFArray,
        base::{CFType, FromVoid, TCFType},
        dictionary::CFDictionary,
        number::CFNumber,
        string::{CFString, CFStringRef},
    };
    use system_configuration::sys::{
        dynamic_store_copy_specific::SCDynamicStoreCopyProxies, schema_definitions::*,
    };

    fn proxy_dict() -> Option<CFDictionary<CFString, CFType>> {
        // Py_BEGIN_ALLOW_THREADS
        let proxy_dict = unsafe { SCDynamicStoreCopyProxies(std::ptr::null()) };
        // Py_END_ALLOW_THREADS
        if proxy_dict.is_null() {
            None
        } else {
            Some(unsafe { CFDictionary::wrap_under_create_rule(proxy_dict) })
        }
    }

    #[pyfunction]
    fn _get_proxy_settings(vm: &VirtualMachine) -> PyResult<Option<PyDictRef>> {
        let Some(proxy_dict) = proxy_dict() else {
            return Ok(None);
        };

        let result = vm.ctx.new_dict();

        let v = 0
            != proxy_dict
                .find(unsafe { kSCPropNetProxiesExcludeSimpleHostnames })
                .and_then(|v| v.downcast::<CFNumber>())
                .and_then(|v| v.to_i32())
                .unwrap_or(0);
        result.set_item("exclude_simple", vm.ctx.new_bool(v).into(), vm)?;

        if let Some(an_array) = proxy_dict
            .find(unsafe { kSCPropNetProxiesExceptionsList })
            .and_then(|v| v.downcast::<CFArray>())
        {
            let v = an_array
                .into_iter()
                .map(|s| {
                    unsafe { CFType::from_void(*s) }
                        .downcast::<CFString>()
                        .map(|s| {
                            let a_string: std::borrow::Cow<str> = (&s).into();
                            PyStr::from(a_string.into_owned())
                        })
                        .to_pyobject(vm)
                })
                .collect();
            result.set_item("exceptions", vm.ctx.new_tuple(v).into(), vm)?;
        }

        Ok(Some(result))
    }

    #[pyfunction]
    fn _get_proxies(vm: &VirtualMachine) -> PyResult<Option<PyDictRef>> {
        let Some(proxy_dict) = proxy_dict() else {
            return Ok(None);
        };

        let result = vm.ctx.new_dict();

        let set_proxy = |result: &PyDictRef,
                         proto: &str,
                         enabled_key: CFStringRef,
                         host_key: CFStringRef,
                         port_key: CFStringRef|
         -> PyResult<()> {
            let enabled = 0
                != proxy_dict
                    .find(enabled_key)
                    .and_then(|v| v.downcast::<CFNumber>())
                    .and_then(|v| v.to_i32())
                    .unwrap_or(0);
            if enabled {
                if let Some(host) = proxy_dict
                    .find(host_key)
                    .and_then(|v| v.downcast::<CFString>())
                {
                    let h = std::borrow::Cow::<str>::from(&host);
                    let v = if let Some(port) = proxy_dict
                        .find(port_key)
                        .and_then(|v| v.downcast::<CFNumber>())
                        .and_then(|v| v.to_i32())
                    {
                        format!("http://{h}:{port}")
                    } else {
                        format!("http://{h}")
                    };
                    result.set_item(proto, vm.new_pyobj(v), vm)?;
                }
            }
            Ok(())
        };

        unsafe {
            set_proxy(
                &result,
                "http",
                kSCPropNetProxiesHTTPEnable,
                kSCPropNetProxiesHTTPProxy,
                kSCPropNetProxiesHTTPPort,
            )?;
            set_proxy(
                &result,
                "https",
                kSCPropNetProxiesHTTPSEnable,
                kSCPropNetProxiesHTTPSProxy,
                kSCPropNetProxiesHTTPSPort,
            )?;
            set_proxy(
                &result,
                "ftp",
                kSCPropNetProxiesFTPEnable,
                kSCPropNetProxiesFTPProxy,
                kSCPropNetProxiesFTPPort,
            )?;
            set_proxy(
                &result,
                "gopher",
                kSCPropNetProxiesGopherEnable,
                kSCPropNetProxiesGopherProxy,
                kSCPropNetProxiesGopherPort,
            )?;
        }
        Ok(Some(result))
    }
}