hf2q 0.1.23

Pure Rust CLI for converting HuggingFace models to hardware-optimized formats and serving them over an OpenAI-compatible API on Apple Silicon
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
146
147
148
149
150
151
152
153
154
155
156
157
158
159

import gzip
import os
import unittest
from unittest.mock import patch

import httpx

import egress_guard


PUBLIC_V4 = "93.184.216.34"


class CompressedStream(httpx.AsyncByteStream):
    def __init__(self, content: bytes):
        self.content = content

    async def __aiter__(self):
        yield self.content


async def public_resolver(hostname: str, port: int):
    del hostname, port
    return [PUBLIC_V4]


class EgressGuardTests(unittest.IsolatedAsyncioTestCase):
    async def test_rejects_non_public_addresses_and_credentialed_urls(self):
        blocked = (
            "http://127.0.0.1/",
            "http://10.0.0.1/",
            "http://169.254.169.254/latest/meta-data/",
            "http://100.64.0.1/",
            "http://[::1]/",
            "http://[::ffff:127.0.0.1]/",
            "http://[2002:7f00:1::]/",
            "http://[2001:0000:4136:e378:8000:63bf:3fff:fdd2]/",
            "http://[64:ff9b::7f00:1]/",
            "http://user:pass@example.com/",
            "http://localhost/",
            "http://example.com:8123/",
        )
        for url in blocked:
            with self.subTest(url=url), self.assertRaises(egress_guard.UnsafeUrlError):
                await egress_guard.resolve_public_target(url)

    async def test_rejects_hostname_when_any_dns_answer_is_private(self):
        async def mixed_resolver(hostname: str, port: int):
            del hostname, port
            return [PUBLIC_V4, "192.168.1.1"]

        with self.assertRaisesRegex(egress_guard.UnsafeUrlError, "outside the public internet"):
            await egress_guard.resolve_public_target("https://example.com/", mixed_resolver)

    async def test_pins_address_and_preserves_host_and_tls_sni(self):
        observed = {}

        async def handler(request: httpx.Request):
            observed["url"] = str(request.url)
            observed["host"] = request.headers["host"]
            observed["sni"] = request.extensions.get("sni_hostname")
            return httpx.Response(200, text="hello")

        response = await egress_guard.guarded_get(
            "https://example.com/docs?q=1",
            timeout=1,
            resolver=public_resolver,
            transport=httpx.MockTransport(handler),
        )
        self.assertEqual(response.text, "hello")
        self.assertEqual(observed["url"], f"https://{PUBLIC_V4}/docs?q=1")
        self.assertEqual(observed["host"], "example.com")
        self.assertEqual(observed["sni"], "example.com")

    async def test_revalidates_every_redirect_and_blocks_public_to_private(self):
        async def handler(request: httpx.Request):
            return httpx.Response(302, headers={"location": "http://127.0.0.1/admin"})

        with self.assertRaises(egress_guard.UnsafeUrlError):
            await egress_guard.guarded_get(
                "https://example.com/",
                timeout=1,
                resolver=public_resolver,
                transport=httpx.MockTransport(handler),
            )

    async def test_rebinding_on_second_hop_is_rejected(self):
        calls = 0

        async def rebinding_resolver(hostname: str, port: int):
            nonlocal calls
            del hostname, port
            calls += 1
            return [PUBLIC_V4] if calls == 1 else ["127.0.0.1"]

        async def handler(request: httpx.Request):
            return httpx.Response(302, headers={"location": "/next"})

        with self.assertRaises(egress_guard.UnsafeUrlError):
            await egress_guard.guarded_get(
                "https://example.com/",
                timeout=1,
                resolver=rebinding_resolver,
                transport=httpx.MockTransport(handler),
            )
        self.assertEqual(calls, 2)

    async def test_https_to_http_redirect_is_rejected(self):
        async def handler(request: httpx.Request):
            return httpx.Response(302, headers={"location": "http://example.org/"})

        with self.assertRaisesRegex(egress_guard.UnsafeUrlError, "HTTPS-to-HTTP"):
            await egress_guard.guarded_get(
                "https://example.com/",
                timeout=1,
                resolver=public_resolver,
                transport=httpx.MockTransport(handler),
            )

    async def test_automatic_path_disables_environment_proxies(self):
        async def handler(request: httpx.Request):
            return httpx.Response(200, text="ok")

        with (
            patch.dict(os.environ, {"HTTPS_PROXY": "http://127.0.0.1:9999"}),
            patch("egress_guard.httpx.AsyncClient", wraps=httpx.AsyncClient) as client,
        ):
            await egress_guard.guarded_get(
                "https://example.com/",
                timeout=1,
                resolver=public_resolver,
                transport=httpx.MockTransport(handler),
            )
        self.assertIs(client.call_args.kwargs["trust_env"], False)

    async def test_real_streaming_path_decodes_compressed_body_once(self):
        expected = "compressed public page"

        async def handler(request: httpx.Request):
            return httpx.Response(
                200,
                headers={
                    "content-encoding": "gzip",
                    "content-type": "text/plain; charset=utf-8",
                },
                stream=CompressedStream(gzip.compress(expected.encode())),
            )

        response = await egress_guard.guarded_get(
            "https://example.com/",
            timeout=1,
            resolver=public_resolver,
            transport=httpx.MockTransport(handler),
        )
        self.assertEqual(response.text, expected)


if __name__ == "__main__":
    unittest.main(verbosity=2)