|
4 | 4 |
|
5 | 5 | """ |
6 | 6 |
|
| 7 | +from fnmatch import fnmatch |
7 | 8 | from functools import lru_cache |
8 | 9 | from typing import Any |
9 | 10 | from typing import NamedTuple |
|
12 | 13 | from urllib.parse import urlparse |
13 | 14 |
|
14 | 15 | from conda import CondaError |
| 16 | +from conda.base.context import context as conda_context |
15 | 17 | from conda.plugins.types import ChannelAuthBase |
16 | 18 | from requests import PreparedRequest |
17 | 19 | from requests import Response |
@@ -41,7 +43,47 @@ class AnacondaAuthError(CondaError): |
41 | 43 |
|
42 | 44 |
|
43 | 45 | def _load_channel_settings(channel_name: str) -> dict[str, Any]: |
44 | | - return {"auth_domain": "auth.some-domain.com"} |
| 46 | + """Find the correct channel settings from conda's configuration.""" |
| 47 | + # TODO(mattkram): Open conda issue to see if we can pass this into the AuthHandler |
| 48 | + # as part of the plugin protocol. |
| 49 | + |
| 50 | + # Since the conda logic uses a url, we derive a url from the channel name |
| 51 | + # this will not work for multi_channels like "defaults", but we restrict the |
| 52 | + # extra fields we need to URL-based channel_settings, which should be sufficient. |
| 53 | + url = channel_name |
| 54 | + if not url.endswith("/"): |
| 55 | + url += "/" |
| 56 | + |
| 57 | + parsed_url = urlparse(url) |
| 58 | + if not parsed_url.scheme or not parsed_url.netloc: |
| 59 | + return {} |
| 60 | + |
| 61 | + # The following implementation has mostly been copied from conda, with one noted exception. |
| 62 | + # Ideally, we can receive the settings in the plugin instantiation. |
| 63 | + # See: https://github.com/conda/conda/blob/2af8e0f7255e1d06ea0bfcb6076c7427d101feee/conda/gateways/connection/session.py#L91-L112 |
| 64 | + |
| 65 | + # We ensure here if there are duplicates defined, we choose the last one |
| 66 | + channel_settings = {} |
| 67 | + for settings in conda_context.channel_settings: |
| 68 | + channel = settings.get("channel", "") |
| 69 | + if channel == channel_name: |
| 70 | + # First we check for exact match |
| 71 | + channel_settings = settings |
| 72 | + continue |
| 73 | + |
| 74 | + parsed_setting = urlparse(channel) |
| 75 | + |
| 76 | + # We require that the schemes must be identical to prevent downgrade attacks. |
| 77 | + # This includes the case of a scheme-less pattern like "*", which is not allowed. |
| 78 | + if parsed_setting.scheme != parsed_url.scheme: |
| 79 | + continue |
| 80 | + |
| 81 | + url_without_schema = parsed_url.netloc + parsed_url.path |
| 82 | + pattern = parsed_setting.netloc + parsed_setting.path |
| 83 | + if fnmatch(url_without_schema, pattern): |
| 84 | + channel_settings = settings |
| 85 | + |
| 86 | + return channel_settings |
45 | 87 |
|
46 | 88 |
|
47 | 89 | class AnacondaAuthHandler(ChannelAuthBase): |
|
0 commit comments