We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
http: allow reading proxies from the current environment
This was something enabled as default by the requests, but not on aiohttp. We have to explicitly enable it to keep the current behavior. https://docs.aiohttp.org/en/stable/client_advanced.html#proxy-support. Discord context: https://discord.com/channels/485586884165107732/563406153334128681/891230518992052274

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `dvc/fs/http.py`
Content:
```
1 import threading
2 
3 from funcy import cached_property, memoize, wrap_with
4 
5 from dvc import prompt
6 from dvc.path_info import HTTPURLInfo
7 from dvc.scheme import Schemes
8 
9 from .fsspec_wrapper import FSSpecWrapper, NoDirectoriesMixin
10 
11 
12 @wrap_with(threading.Lock())
13 @memoize
14 def ask_password(host, user):
15     return prompt.password(
16         "Enter a password for "
17         "host '{host}' user '{user}'".format(host=host, user=user)
18     )
19 
20 
21 def make_context(ssl_verify):
22     if isinstance(ssl_verify, bool) or ssl_verify is None:
23         return ssl_verify
24 
25     # If this is a path, then we will create an
26     # SSL context for it, and load the given certificate.
27     import ssl
28 
29     context = ssl.create_default_context()
30     context.load_verify_locations(ssl_verify)
31     return context
32 
33 
34 # pylint: disable=abstract-method
35 class HTTPFileSystem(NoDirectoriesMixin, FSSpecWrapper):
36     scheme = Schemes.HTTP
37     PATH_CLS = HTTPURLInfo
38     PARAM_CHECKSUM = "checksum"
39     REQUIRES = {"aiohttp": "aiohttp", "aiohttp-retry": "aiohttp_retry"}
40     CAN_TRAVERSE = False
41 
42     SESSION_RETRIES = 5
43     SESSION_BACKOFF_FACTOR = 0.1
44     REQUEST_TIMEOUT = 60
45 
46     def _prepare_credentials(self, **config):
47         import aiohttp
48         from fsspec.asyn import fsspec_loop
49 
50         from dvc.config import ConfigError
51 
52         credentials = {}
53         client_kwargs = credentials.setdefault("client_kwargs", {})
54 
55         if config.get("auth"):
56             user = config.get("user")
57             password = config.get("password")
58             custom_auth_header = config.get("custom_auth_header")
59 
60             if password is None and config.get("ask_password"):
61                 password = ask_password(config.get("url"), user or "custom")
62 
63             auth_method = config["auth"]
64             if auth_method == "basic":
65                 if user is None or password is None:
66                     raise ConfigError(
67                         "HTTP 'basic' authentication require both "
68                         "'user' and 'password'"
69                     )
70 
71                 client_kwargs["auth"] = aiohttp.BasicAuth(user, password)
72             elif auth_method == "custom":
73                 if custom_auth_header is None or password is None:
74                     raise ConfigError(
75                         "HTTP 'custom' authentication require both "
76                         "'custom_auth_header' and 'password'"
77                     )
78                 credentials["headers"] = {custom_auth_header: password}
79             else:
80                 raise NotImplementedError(
81                     f"Auth method {auth_method!r} is not supported."
82                 )
83 
84         if "ssl_verify" in config:
85             with fsspec_loop():
86                 client_kwargs["connector"] = aiohttp.TCPConnector(
87                     ssl=make_context(config["ssl_verify"])
88                 )
89 
90         credentials["get_client"] = self.get_client
91         self.upload_method = config.get("method", "POST")
92         return credentials
93 
94     async def get_client(self, **kwargs):
95         import aiohttp
96         from aiohttp_retry import ExponentialRetry, RetryClient
97 
98         kwargs["retry_options"] = ExponentialRetry(
99             attempts=self.SESSION_RETRIES,
100             factor=self.SESSION_BACKOFF_FACTOR,
101             max_timeout=self.REQUEST_TIMEOUT,
102         )
103 
104         # The default timeout for the aiohttp is 300 seconds
105         # which is too low for DVC's interactions (especially
106         # on the read) when dealing with large data blobs. We
107         # unlimit the total time to read, and only limit the
108         # time that is spent when connecting to the remote server.
109         kwargs["timeout"] = aiohttp.ClientTimeout(
110             total=None,
111             connect=self.REQUEST_TIMEOUT,
112             sock_connect=self.REQUEST_TIMEOUT,
113             sock_read=None,
114         )
115 
116         return RetryClient(**kwargs)
117 
118     @cached_property
119     def fs(self):
120         from fsspec.implementations.http import (
121             HTTPFileSystem as _HTTPFileSystem,
122         )
123 
124         return _HTTPFileSystem(**self.fs_args)
125 
126     def _entry_hook(self, entry):
127         entry["checksum"] = entry.get("ETag") or entry.get("Content-MD5")
128         return entry
129
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/dvc/fs/http.py b/dvc/fs/http.py
--- a/dvc/fs/http.py
+++ b/dvc/fs/http.py
@@ -87,6 +87,9 @@
                     ssl=make_context(config["ssl_verify"])
                 )
 
+        # Allow reading proxy configurations from the environment.
+        client_kwargs["trust_env"] = True
+
         credentials["get_client"] = self.get_client
         self.upload_method = config.get("method", "POST")
         return credentials

{"golden_diff": "diff --git a/dvc/fs/http.py b/dvc/fs/http.py\n--- a/dvc/fs/http.py\n+++ b/dvc/fs/http.py\n@@ -87,6 +87,9 @@\n                     ssl=make_context(config[\"ssl_verify\"])\n                 )\n \n+        # Allow reading proxy configurations from the environment.\n+        client_kwargs[\"trust_env\"] = True\n+\n         credentials[\"get_client\"] = self.get_client\n         self.upload_method = config.get(\"method\", \"POST\")\n         return credentials\n", "issue": "http: allow reading proxies from the current environment\nThis was something enabled as default by the requests, but not on aiohttp. We have to explicitly enable it to keep the current behavior. https://docs.aiohttp.org/en/stable/client_advanced.html#proxy-support. Discord context: https://discord.com/channels/485586884165107732/563406153334128681/891230518992052274\n", "before_files": [{"content": "import threading\n\nfrom funcy import cached_property, memoize, wrap_with\n\nfrom dvc import prompt\nfrom dvc.path_info import HTTPURLInfo\nfrom dvc.scheme import Schemes\n\nfrom .fsspec_wrapper import FSSpecWrapper, NoDirectoriesMixin\n\n\n@wrap_with(threading.Lock())\n@memoize\ndef ask_password(host, user):\n    return prompt.password(\n        \"Enter a password for \"\n        \"host '{host}' user '{user}'\".format(host=host, user=user)\n    )\n\n\ndef make_context(ssl_verify):\n    if isinstance(ssl_verify, bool) or ssl_verify is None:\n        return ssl_verify\n\n    # If this is a path, then we will create an\n    # SSL context for it, and load the given certificate.\n    import ssl\n\n    context = ssl.create_default_context()\n    context.load_verify_locations(ssl_verify)\n    return context\n\n\n# pylint: disable=abstract-method\nclass HTTPFileSystem(NoDirectoriesMixin, FSSpecWrapper):\n    scheme = Schemes.HTTP\n    PATH_CLS = HTTPURLInfo\n    PARAM_CHECKSUM = \"checksum\"\n    REQUIRES = {\"aiohttp\": \"aiohttp\", \"aiohttp-retry\": \"aiohttp_retry\"}\n    CAN_TRAVERSE = False\n\n    SESSION_RETRIES = 5\n    SESSION_BACKOFF_FACTOR = 0.1\n    REQUEST_TIMEOUT = 60\n\n    def _prepare_credentials(self, **config):\n        import aiohttp\n        from fsspec.asyn import fsspec_loop\n\n        from dvc.config import ConfigError\n\n        credentials = {}\n        client_kwargs = credentials.setdefault(\"client_kwargs\", {})\n\n        if config.get(\"auth\"):\n            user = config.get(\"user\")\n            password = config.get(\"password\")\n            custom_auth_header = config.get(\"custom_auth_header\")\n\n            if password is None and config.get(\"ask_password\"):\n                password = ask_password(config.get(\"url\"), user or \"custom\")\n\n            auth_method = config[\"auth\"]\n            if auth_method == \"basic\":\n                if user is None or password is None:\n                    raise ConfigError(\n                        \"HTTP 'basic' authentication require both \"\n                        \"'user' and 'password'\"\n                    )\n\n                client_kwargs[\"auth\"] = aiohttp.BasicAuth(user, password)\n            elif auth_method == \"custom\":\n                if custom_auth_header is None or password is None:\n                    raise ConfigError(\n                        \"HTTP 'custom' authentication require both \"\n                        \"'custom_auth_header' and 'password'\"\n                    )\n                credentials[\"headers\"] = {custom_auth_header: password}\n            else:\n                raise NotImplementedError(\n                    f\"Auth method {auth_method!r} is not supported.\"\n                )\n\n        if \"ssl_verify\" in config:\n            with fsspec_loop():\n                client_kwargs[\"connector\"] = aiohttp.TCPConnector(\n                    ssl=make_context(config[\"ssl_verify\"])\n                )\n\n        credentials[\"get_client\"] = self.get_client\n        self.upload_method = config.get(\"method\", \"POST\")\n        return credentials\n\n    async def get_client(self, **kwargs):\n        import aiohttp\n        from aiohttp_retry import ExponentialRetry, RetryClient\n\n        kwargs[\"retry_options\"] = ExponentialRetry(\n            attempts=self.SESSION_RETRIES,\n            factor=self.SESSION_BACKOFF_FACTOR,\n            max_timeout=self.REQUEST_TIMEOUT,\n        )\n\n        # The default timeout for the aiohttp is 300 seconds\n        # which is too low for DVC's interactions (especially\n        # on the read) when dealing with large data blobs. We\n        # unlimit the total time to read, and only limit the\n        # time that is spent when connecting to the remote server.\n        kwargs[\"timeout\"] = aiohttp.ClientTimeout(\n            total=None,\n            connect=self.REQUEST_TIMEOUT,\n            sock_connect=self.REQUEST_TIMEOUT,\n            sock_read=None,\n        )\n\n        return RetryClient(**kwargs)\n\n    @cached_property\n    def fs(self):\n        from fsspec.implementations.http import (\n            HTTPFileSystem as _HTTPFileSystem,\n        )\n\n        return _HTTPFileSystem(**self.fs_args)\n\n    def _entry_hook(self, entry):\n        entry[\"checksum\"] = entry.get(\"ETag\") or entry.get(\"Content-MD5\")\n        return entry\n", "path": "dvc/fs/http.py"}], "after_files": [{"content": "import threading\n\nfrom funcy import cached_property, memoize, wrap_with\n\nfrom dvc import prompt\nfrom dvc.path_info import HTTPURLInfo\nfrom dvc.scheme import Schemes\n\nfrom .fsspec_wrapper import FSSpecWrapper, NoDirectoriesMixin\n\n\n@wrap_with(threading.Lock())\n@memoize\ndef ask_password(host, user):\n    return prompt.password(\n        \"Enter a password for \"\n        \"host '{host}' user '{user}'\".format(host=host, user=user)\n    )\n\n\ndef make_context(ssl_verify):\n    if isinstance(ssl_verify, bool) or ssl_verify is None:\n        return ssl_verify\n\n    # If this is a path, then we will create an\n    # SSL context for it, and load the given certificate.\n    import ssl\n\n    context = ssl.create_default_context()\n    context.load_verify_locations(ssl_verify)\n    return context\n\n\n# pylint: disable=abstract-method\nclass HTTPFileSystem(NoDirectoriesMixin, FSSpecWrapper):\n    scheme = Schemes.HTTP\n    PATH_CLS = HTTPURLInfo\n    PARAM_CHECKSUM = \"checksum\"\n    REQUIRES = {\"aiohttp\": \"aiohttp\", \"aiohttp-retry\": \"aiohttp_retry\"}\n    CAN_TRAVERSE = False\n\n    SESSION_RETRIES = 5\n    SESSION_BACKOFF_FACTOR = 0.1\n    REQUEST_TIMEOUT = 60\n\n    def _prepare_credentials(self, **config):\n        import aiohttp\n        from fsspec.asyn import fsspec_loop\n\n        from dvc.config import ConfigError\n\n        credentials = {}\n        client_kwargs = credentials.setdefault(\"client_kwargs\", {})\n\n        if config.get(\"auth\"):\n            user = config.get(\"user\")\n            password = config.get(\"password\")\n            custom_auth_header = config.get(\"custom_auth_header\")\n\n            if password is None and config.get(\"ask_password\"):\n                password = ask_password(config.get(\"url\"), user or \"custom\")\n\n            auth_method = config[\"auth\"]\n            if auth_method == \"basic\":\n                if user is None or password is None:\n                    raise ConfigError(\n                        \"HTTP 'basic' authentication require both \"\n                        \"'user' and 'password'\"\n                    )\n\n                client_kwargs[\"auth\"] = aiohttp.BasicAuth(user, password)\n            elif auth_method == \"custom\":\n                if custom_auth_header is None or password is None:\n                    raise ConfigError(\n                        \"HTTP 'custom' authentication require both \"\n                        \"'custom_auth_header' and 'password'\"\n                    )\n                credentials[\"headers\"] = {custom_auth_header: password}\n            else:\n                raise NotImplementedError(\n                    f\"Auth method {auth_method!r} is not supported.\"\n                )\n\n        if \"ssl_verify\" in config:\n            with fsspec_loop():\n                client_kwargs[\"connector\"] = aiohttp.TCPConnector(\n                    ssl=make_context(config[\"ssl_verify\"])\n                )\n\n        # Allow reading proxy configurations from the environment.\n        client_kwargs[\"trust_env\"] = True\n\n        credentials[\"get_client\"] = self.get_client\n        self.upload_method = config.get(\"method\", \"POST\")\n        return credentials\n\n    async def get_client(self, **kwargs):\n        import aiohttp\n        from aiohttp_retry import ExponentialRetry, RetryClient\n\n        kwargs[\"retry_options\"] = ExponentialRetry(\n            attempts=self.SESSION_RETRIES,\n            factor=self.SESSION_BACKOFF_FACTOR,\n            max_timeout=self.REQUEST_TIMEOUT,\n        )\n\n        # The default timeout for the aiohttp is 300 seconds\n        # which is too low for DVC's interactions (especially\n        # on the read) when dealing with large data blobs. We\n        # unlimit the total time to read, and only limit the\n        # time that is spent when connecting to the remote server.\n        kwargs[\"timeout\"] = aiohttp.ClientTimeout(\n            total=None,\n            connect=self.REQUEST_TIMEOUT,\n            sock_connect=self.REQUEST_TIMEOUT,\n            sock_read=None,\n        )\n\n        return RetryClient(**kwargs)\n\n    @cached_property\n    def fs(self):\n        from fsspec.implementations.http import (\n            HTTPFileSystem as _HTTPFileSystem,\n        )\n\n        return _HTTPFileSystem(**self.fs_args)\n\n    def _entry_hook(self, entry):\n        entry[\"checksum\"] = entry.get(\"ETag\") or entry.get(\"Content-MD5\")\n        return entry\n", "path": "dvc/fs/http.py"}]}