|
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 Optional |
@@ -39,18 +40,38 @@ def _load_settings_for_channel(channel_name: str) -> dict[str, Any]: |
39 | 40 | """Find the correct channel settings from conda's configuration.""" |
40 | 41 | # TODO(mattkram): Open conda issue to see if we can pass this into the AuthHandler |
41 | 42 | # as part of the plugin protocol. |
42 | | - for settings in global_context.channel_settings: |
43 | | - settings_channel = settings.get("channel") |
44 | 43 |
|
45 | | - # TODO(mattkram): This is not robust as it assumes the glob pattern |
46 | | - if settings_channel.endswith("/*"): |
47 | | - prefix = settings_channel[:-2] |
48 | | - if channel_name.startswith(prefix): |
49 | | - return settings |
| 44 | + # The following implementation has been copied from conda. Ideally, we can receive |
| 45 | + # the settings in the plugin instantiation. |
50 | 46 |
|
51 | | - # TODO(mattkram): How should we handle this? |
52 | | - return {} |
53 | | - raise ValueError(f"Couldn't find the settings for channel {channel_name}") |
| 47 | + # We ensure here if there are duplicates defined, we choose the last one |
| 48 | + channel_settings = {} |
| 49 | + for settings in global_context.channel_settings: |
| 50 | + channel = settings.get("channel", "") |
| 51 | + if channel == channel_name: |
| 52 | + # First we check for exact match |
| 53 | + channel_settings = settings |
| 54 | + continue |
| 55 | + |
| 56 | + # If we don't have an exact match, we attempt to match a URL pattern |
| 57 | + # The following was adjusted since we have channel name and not a URL |
| 58 | + parsed_url = urlparse(channel_name) |
| 59 | + if not parsed_url.scheme or not parsed_url.netloc: |
| 60 | + continue |
| 61 | + |
| 62 | + parsed_setting = urlparse(channel) |
| 63 | + |
| 64 | + # We require that the schemes must be identical to prevent downgrade attacks. |
| 65 | + # This includes the case of a scheme-less pattern like "*", which is not allowed. |
| 66 | + if parsed_setting.scheme != parsed_url.scheme: |
| 67 | + continue |
| 68 | + |
| 69 | + url_without_schema = parsed_url.netloc + parsed_url.path |
| 70 | + pattern = parsed_setting.netloc + parsed_setting.path |
| 71 | + if fnmatch(url_without_schema, pattern): |
| 72 | + channel_settings = settings |
| 73 | + |
| 74 | + return channel_settings |
54 | 75 |
|
55 | 76 |
|
56 | 77 | class AnacondaAuthHandlerOverrides(BaseModel): |
|
0 commit comments