mirrors: distinguish between source/binary mirror; simplify schema (#34523)

Allow the following formats:

```yaml
mirrors:
  name: <url>
```

```yaml
mirrors:
  name:
    url: s3://xyz
    access_pair: [x, y]
```

```yaml
mirrors:
  name:
    fetch: http://xyz
    push:
      url: s3://xyz
      access_pair: [x, y]
```

And reserve two new properties to indicate the mirror type (e.g.
mirror.spack.io is a source mirror, not a binary cache)

```yaml
mirrors:
  spack-public:
    source: true
    binary: false
    url: https://mirror.spack.io
```

13 files changed, 484 insertions, 302 deletions

diff --git a/etc/spack/defaults/mirrors.yaml b/etc/spack/defaults/mirrors.yaml
index 4db4f7dedb..0891ae4504 100644
--- a/etc/spack/defaults/mirrors.yaml
+++ b/etc/spack/defaults/mirrors.yaml
@@ -1,2 +1,4 @@
 mirrors:
-  spack-public: https://mirror.spack.io
+  spack-public:
+    binary: false
+    url: https://mirror.spack.io
diff --git a/lib/spack/spack/binary_distribution.py b/lib/spack/spack/binary_distribution.py
index 53884a2912..6f6e0de4a3 100644
--- a/lib/spack/spack/binary_distribution.py
+++ b/lib/spack/spack/binary_distribution.py
@@ -317,9 +317,9 @@ class BinaryCacheIndex:
         from each configured mirror and stored locally (both in memory and
         on disk under ``_index_cache_root``)."""
         self._init_local_index_cache()
-
-        mirrors = spack.mirror.MirrorCollection()
-        configured_mirror_urls = [m.fetch_url for m in mirrors.values()]
+        configured_mirror_urls = [
+            m.fetch_url for m in spack.mirror.MirrorCollection(binary=True).values()
+        ]
         items_to_remove = []
         spec_cache_clear_needed = False
         spec_cache_regenerate_needed = not self._mirrors_for_spec
@@ -1465,8 +1465,9 @@ def download_tarball(spec, unsigned=False, mirrors_for_spec=None):
            "signature_verified": "true-if-binary-pkg-was-already-verified"
        }
     """
-    if not spack.mirror.MirrorCollection():
-        tty.die("Please add a spack mirror to allow " + "download of pre-compiled packages.")
+    configured_mirrors = spack.mirror.MirrorCollection(binary=True).values()
+    if not configured_mirrors:
+        tty.die("Please add a spack mirror to allow download of pre-compiled packages.")
 
     tarball = tarball_path_name(spec, ".spack")
     specfile_prefix = tarball_name(spec, ".spec")
@@ -1483,11 +1484,7 @@ def download_tarball(spec, unsigned=False, mirrors_for_spec=None):
     # we need was in an un-indexed mirror.  No need to check any
     # mirror for the spec twice though.
     try_first = [i["mirror_url"] for i in mirrors_for_spec] if mirrors_for_spec else []
-    try_next = [
-        i.fetch_url
-        for i in spack.mirror.MirrorCollection().values()
-        if i.fetch_url not in try_first
-    ]
+    try_next = [i.fetch_url for i in configured_mirrors if i.fetch_url not in try_first]
 
     for url in try_first + try_next:
         mirrors_to_try.append(
@@ -1980,7 +1977,9 @@ def try_direct_fetch(spec, mirrors=None):
     specfile_is_signed = False
     found_specs = []
 
-    for mirror in spack.mirror.MirrorCollection(mirrors=mirrors).values():
+    binary_mirrors = spack.mirror.MirrorCollection(mirrors=mirrors, binary=True).values()
+
+    for mirror in binary_mirrors:
         buildcache_fetch_url_json = url_util.join(
             mirror.fetch_url, _build_cache_relative_path, specfile_name
         )
@@ -2043,7 +2042,7 @@ def get_mirrors_for_spec(spec=None, mirrors_to_check=None, index_only=False):
     if spec is None:
         return []
 
-    if not spack.mirror.MirrorCollection(mirrors=mirrors_to_check):
+    if not spack.mirror.MirrorCollection(mirrors=mirrors_to_check, binary=True):
         tty.debug("No Spack mirrors are currently configured")
         return {}
 
@@ -2082,7 +2081,7 @@ def clear_spec_cache():
 
 def get_keys(install=False, trust=False, force=False, mirrors=None):
     """Get pgp public keys available on mirror with suffix .pub"""
-    mirror_collection = mirrors or spack.mirror.MirrorCollection()
+    mirror_collection = mirrors or spack.mirror.MirrorCollection(binary=True)
 
     if not mirror_collection:
         tty.die("Please add a spack mirror to allow " + "download of build caches.")
@@ -2243,7 +2242,7 @@ def check_specs_against_mirrors(mirrors, specs, output_file=None):
 
     """
     rebuilds = {}
-    for mirror in spack.mirror.MirrorCollection(mirrors).values():
+    for mirror in spack.mirror.MirrorCollection(mirrors, binary=True).values():
         tty.debug("Checking for built specs at {0}".format(mirror.fetch_url))
 
         rebuild_list = []
@@ -2287,7 +2286,7 @@ def _download_buildcache_entry(mirror_root, descriptions):
 
 
 def download_buildcache_entry(file_descriptions, mirror_url=None):
-    if not mirror_url and not spack.mirror.MirrorCollection():
+    if not mirror_url and not spack.mirror.MirrorCollection(binary=True):
         tty.die(
             "Please provide or add a spack mirror to allow " + "download of buildcache entries."
         )
@@ -2296,7 +2295,7 @@ def download_buildcache_entry(file_descriptions, mirror_url=None):
         mirror_root = os.path.join(mirror_url, _build_cache_relative_path)
         return _download_buildcache_entry(mirror_root, file_descriptions)
 
-    for mirror in spack.mirror.MirrorCollection().values():
+    for mirror in spack.mirror.MirrorCollection(binary=True).values():
         mirror_root = os.path.join(mirror.fetch_url, _build_cache_relative_path)
 
         if _download_buildcache_entry(mirror_root, file_descriptions):
diff --git a/lib/spack/spack/ci.py b/lib/spack/spack/ci.py
index cfe87e5214..fd0f5fe20f 100644
--- a/lib/spack/spack/ci.py
+++ b/lib/spack/spack/ci.py
@@ -224,7 +224,7 @@ def _print_staging_summary(spec_labels, stages, mirrors_to_check, rebuild_decisi
     if not stages:
         return
 
-    mirrors = spack.mirror.MirrorCollection(mirrors=mirrors_to_check)
+    mirrors = spack.mirror.MirrorCollection(mirrors=mirrors_to_check, binary=True)
     tty.msg("Checked the following mirrors for binaries:")
     for m in mirrors.values():
         tty.msg("  {0}".format(m.fetch_url))
diff --git a/lib/spack/spack/cmd/gpg.py b/lib/spack/spack/cmd/gpg.py
index 6192d25bd7..0bce5f4736 100644
--- a/lib/spack/spack/cmd/gpg.py
+++ b/lib/spack/spack/cmd/gpg.py
@@ -216,7 +216,7 @@ def gpg_publish(args):
         url = spack.util.url.path_to_file_url(args.directory)
         mirror = spack.mirror.Mirror(url, url)
     elif args.mirror_name:
-        mirror = spack.mirror.MirrorCollection().lookup(args.mirror_name)
+        mirror = spack.mirror.MirrorCollection(binary=True).lookup(args.mirror_name)
     elif args.mirror_url:
         mirror = spack.mirror.Mirror(args.mirror_url, args.mirror_url)
 
diff --git a/lib/spack/spack/cmd/mirror.py b/lib/spack/spack/cmd/mirror.py
index f8a84e9af2..4b853c1bc0 100644
--- a/lib/spack/spack/cmd/mirror.py
+++ b/lib/spack/spack/cmd/mirror.py
@@ -21,7 +21,6 @@ import spack.spec
 import spack.util.path
 import spack.util.web as web_util
 from spack.error import SpackError
-from spack.util.spack_yaml import syaml_dict
 
 description = "manage mirrors (source and binary)"
 section = "config"
@@ -104,6 +103,15 @@ def setup_parser(subparser):
         default=spack.config.default_modify_scope(),
         help="configuration scope to modify",
     )
+    add_parser.add_argument(
+        "--type",
+        action="append",
+        choices=("binary", "source"),
+        help=(
+            "specify the mirror type: for both binary "
+            "and source use `--type binary --type source` (default)"
+        ),
+    )
     arguments.add_s3_connection_args(add_parser, False)
     # Remove
     remove_parser = sp.add_parser("remove", aliases=["rm"], help=mirror_remove.__doc__)
@@ -120,8 +128,12 @@ def setup_parser(subparser):
     set_url_parser = sp.add_parser("set-url", help=mirror_set_url.__doc__)
     set_url_parser.add_argument("name", help="mnemonic name for mirror", metavar="mirror")
     set_url_parser.add_argument("url", help="url of mirror directory from 'spack mirror create'")
-    set_url_parser.add_argument(
-        "--push", action="store_true", help="set only the URL used for uploading new packages"
+    set_url_push_or_fetch = set_url_parser.add_mutually_exclusive_group(required=False)
+    set_url_push_or_fetch.add_argument(
+        "--push", action="store_true", help="set only the URL used for uploading"
+    )
+    set_url_push_or_fetch.add_argument(
+        "--fetch", action="store_true", help="set only the URL used for downloading"
     )
     set_url_parser.add_argument(
         "--scope",
@@ -132,6 +144,35 @@ def setup_parser(subparser):
     )
     arguments.add_s3_connection_args(set_url_parser, False)
 
+    # Set
+    set_parser = sp.add_parser("set", help=mirror_set.__doc__)
+    set_parser.add_argument("name", help="mnemonic name for mirror", metavar="mirror")
+    set_parser_push_or_fetch = set_parser.add_mutually_exclusive_group(required=False)
+    set_parser_push_or_fetch.add_argument(
+        "--push", action="store_true", help="modify just the push connection details"
+    )
+    set_parser_push_or_fetch.add_argument(
+        "--fetch", action="store_true", help="modify just the fetch connection details"
+    )
+    set_parser.add_argument(
+        "--type",
+        action="append",
+        choices=("binary", "source"),
+        help=(
+            "specify the mirror type: for both binary "
+            "and source use `--type binary --type source`"
+        ),
+    )
+    set_parser.add_argument("--url", help="url of mirror directory from 'spack mirror create'")
+    set_parser.add_argument(
+        "--scope",
+        choices=scopes,
+        metavar=scopes_metavar,
+        default=spack.config.default_modify_scope(),
+        help="configuration scope to modify",
+    )
+    arguments.add_s3_connection_args(set_parser, False)
+
     # List
     list_parser = sp.add_parser("list", help=mirror_list.__doc__)
     list_parser.add_argument(
@@ -151,17 +192,21 @@ def mirror_add(args):
         or args.s3_access_token
         or args.s3_profile
         or args.s3_endpoint_url
+        or args.type
     ):
         connection = {"url": args.url}
         if args.s3_access_key_id and args.s3_access_key_secret:
-            connection["access_pair"] = (args.s3_access_key_id, args.s3_access_key_secret)
+            connection["access_pair"] = [args.s3_access_key_id, args.s3_access_key_secret]
         if args.s3_access_token:
             connection["access_token"] = args.s3_access_token
         if args.s3_profile:
             connection["profile"] = args.s3_profile
         if args.s3_endpoint_url:
             connection["endpoint_url"] = args.s3_endpoint_url
-        mirror = spack.mirror.Mirror(fetch_url=connection, push_url=connection, name=args.name)
+        if args.type:
+            connection["binary"] = "binary" in args.type
+            connection["source"] = "source" in args.type
+        mirror = spack.mirror.Mirror(connection, name=args.name)
     else:
         mirror = spack.mirror.Mirror(args.url, name=args.name)
     spack.mirror.add(mirror, args.scope)
@@ -172,73 +217,49 @@ def mirror_remove(args):
     spack.mirror.remove(args.name, args.scope)
 
 
-def mirror_set_url(args):
-    """change the URL of a mirror"""
-    url = args.url
+def _configure_mirror(args):
     mirrors = spack.config.get("mirrors", scope=args.scope)
-    if not mirrors:
-        mirrors = syaml_dict()
 
     if args.name not in mirrors:
-        tty.die("No mirror found with name %s." % args.name)
-
-    entry = mirrors[args.name]
-    key_values = ["s3_access_key_id", "s3_access_token", "s3_profile"]
-
-    if any(value for value in key_values if value in args):
-        incoming_data = {
-            "url": url,
-            "access_pair": (args.s3_access_key_id, args.s3_access_key_secret),
-            "access_token": args.s3_access_token,
-            "profile": args.s3_profile,
-            "endpoint_url": args.s3_endpoint_url,
-        }
-    try:
-        fetch_url = entry["fetch"]
-        push_url = entry["push"]
-    except TypeError:
-        fetch_url, push_url = entry, entry
-
-    changes_made = False
-
-    if args.push:
-        if isinstance(push_url, dict):
-            changes_made = changes_made or push_url != incoming_data
-            push_url = incoming_data
-        else:
-            changes_made = changes_made or push_url != url
-            push_url = url
+        tty.die(f"No mirror found with name {args.name}.")
+
+    entry = spack.mirror.Mirror(mirrors[args.name], args.name)
+    direction = "fetch" if args.fetch else "push" if args.push else None
+    changes = {}
+    if args.url:
+        changes["url"] = args.url
+    if args.s3_access_key_id and args.s3_access_key_secret:
+        changes["access_pair"] = [args.s3_access_key_id, args.s3_access_key_secret]
+    if args.s3_access_token:
+        changes["access_token"] = args.s3_access_token
+    if args.s3_profile:
+        changes["profile"] = args.s3_profile
+    if args.s3_endpoint_url:
+        changes["endpoint_url"] = args.s3_endpoint_url
+
+    # argparse cannot distinguish between --binary and --no-binary when same dest :(
+    # notice that set-url does not have these args, so getattr
+    if getattr(args, "type", None):
+        changes["binary"] = "binary" in args.type
+        changes["source"] = "source" in args.type
+
+    changed = entry.update(changes, direction)
+
+    if changed:
+        mirrors[args.name] = entry.to_dict()
+        spack.config.set("mirrors", mirrors, scope=args.scope)
     else:
-        if isinstance(push_url, dict):
-            changes_made = changes_made or push_url != incoming_data or push_url != incoming_data
-            fetch_url, push_url = incoming_data, incoming_data
-        else:
-            changes_made = changes_made or push_url != url
-            fetch_url, push_url = url, url
-
-    items = [
-        (
-            (n, u)
-            if n != args.name
-            else (
-                (n, {"fetch": fetch_url, "push": push_url})
-                if fetch_url != push_url
-                else (n, {"fetch": fetch_url, "push": fetch_url})
-            )
-        )
-        for n, u in mirrors.items()
-    ]
+        tty.msg("No changes made to mirror %s." % args.name)
 
-    mirrors = syaml_dict(items)
-    spack.config.set("mirrors", mirrors, scope=args.scope)
 
-    if changes_made:
-        tty.msg(
-            "Changed%s url or connection information for mirror %s."
-            % ((" (push)" if args.push else ""), args.name)
-        )
-    else:
-        tty.msg("No changes made to mirror %s." % args.name)
+def mirror_set(args):
+    """Configure the connection details of a mirror"""
+    _configure_mirror(args)
+
+
+def mirror_set_url(args):
+    """Change the URL of a mirror."""
+    _configure_mirror(args)
 
 
 def mirror_list(args):
@@ -488,6 +509,7 @@ def mirror(parser, args):
         "remove": mirror_remove,
         "rm": mirror_remove,
         "set-url": mirror_set_url,
+        "set": mirror_set,
         "list": mirror_list,
     }
 
diff --git a/lib/spack/spack/installer.py b/lib/spack/spack/installer.py
index 015e244f98..3b8e76c101 100644
--- a/lib/spack/spack/installer.py
+++ b/lib/spack/spack/installer.py
@@ -505,8 +505,8 @@ def _try_install_from_binary_cache(
             otherwise, ``False``
         timer: timer to keep track of binary install phases.
     """
-    # Early exit if no mirrors are configured.
-    if not spack.mirror.MirrorCollection():
+    # Early exit if no binary mirrors are configured.
+    if not spack.mirror.MirrorCollection(binary=True):
         return False
 
     tty.debug("Searching for binary cache of {0}".format(package_id(pkg)))
diff --git a/lib/spack/spack/mirror.py b/lib/spack/spack/mirror.py
index 65a1379455..d6be146291 100644
--- a/lib/spack/spack/mirror.py
+++ b/lib/spack/spack/mirror.py
@@ -18,6 +18,7 @@ import os.path
 import sys
 import traceback
 import urllib.parse
+from typing import Optional, Union
 
 import llnl.util.tty as tty
 from llnl.util.filesystem import mkdirp
@@ -40,15 +41,6 @@ from spack.version import VersionList
 supported_url_schemes = ("file", "http", "https", "sftp", "ftp", "s3", "gs")
 
 
-def _display_mirror_entry(size, name, url, type_=None):
-    if type_:
-        type_ = "".join((" (", type_, ")"))
-    else:
-        type_ = ""
-
-    print("%-*s%s%s" % (size + 4, name, url, type_))
-
-
 def _url_or_path_to_url(url_or_path: str) -> str:
     """For simplicity we allow mirror URLs in config files to be local, relative paths.
     This helper function takes care of distinguishing between URLs and paths, and
@@ -71,36 +63,24 @@ class Mirror:
     to them. These two URLs are usually the same.
     """
 
-    def __init__(self, fetch_url, push_url=None, name=None):
-        self._fetch_url = fetch_url
-        self._push_url = push_url
+    def __init__(self, data: Union[str, dict], name: Optional[str] = None):
+        self._data = data
         self._name = name
 
-    def __eq__(self, other):
-        return self._fetch_url == other._fetch_url and self._push_url == other._push_url
-
-    def to_json(self, stream=None):
-        return sjson.dump(self.to_dict(), stream)
-
-    def to_yaml(self, stream=None):
-        return syaml.dump(self.to_dict(), stream)
-
     @staticmethod
     def from_yaml(stream, name=None):
-        data = syaml.load(stream)
-        return Mirror.from_dict(data, name)
+        return Mirror(syaml.load(stream), name)
 
     @staticmethod
     def from_json(stream, name=None):
         try:
-            d = sjson.load(stream)
-            return Mirror.from_dict(d, name)
+            return Mirror(sjson.load(stream), name)
         except Exception as e:
             raise sjson.SpackJSONError("error parsing JSON mirror:", str(e)) from e
 
     @staticmethod
     def from_local_path(path: str):
-        return Mirror(fetch_url=url_util.path_to_file_url(path))
+        return Mirror(url_util.path_to_file_url(path))
 
     @staticmethod
     def from_url(url: str):
@@ -111,165 +91,220 @@ class Mirror:
                     url, ", ".join(supported_url_schemes)
                 )
             )
-        return Mirror(fetch_url=url)
+        return Mirror(url)
 
-    def to_dict(self):
-        # Keep it a key-value pair <name>: <url> when possible.
-        if isinstance(self._fetch_url, str) and self._push_url is None:
-            return self._fetch_url
-
-        if self._push_url is None:
-            return syaml_dict([("fetch", self._fetch_url), ("push", self._fetch_url)])
-        else:
-            return syaml_dict([("fetch", self._fetch_url), ("push", self._push_url)])
+    def __eq__(self, other):
+        if not isinstance(other, Mirror):
+            return NotImplemented
+        return self._data == other._data and self._name == other._name
 
-    @staticmethod
-    def from_dict(d, name=None):
-        if isinstance(d, str):
-            return Mirror(d, name=name)
-        else:
-            return Mirror(d["fetch"], d["push"], name=name)
+    def __str__(self):
+        return f"{self._name}: {self.push_url} {self.fetch_url}"
 
-    def display(self, max_len=0):
-        if self._push_url is None:
-            _display_mirror_entry(max_len, self._name, self.fetch_url)
-        else:
-            _display_mirror_entry(max_len, self._name, self.fetch_url, "fetch")
-            _display_mirror_entry(max_len, self._name, self.push_url, "push")
+    def __repr__(self):
+        return f"Mirror(name={self._name!r}, data={self._data!r})"
 
-    def __str__(self):
-        name = self._name
-        if name is None:
-            name = ""
-        else:
-            name = ' "%s"' % name
+    def to_json(self, stream=None):
+        return sjson.dump(self.to_dict(), stream)
 
-        if self._push_url is None:
-            return "[Mirror%s (%s)]" % (name, self._fetch_url)
+    def to_yaml(self, stream=None):
+        return syaml.dump(self.to_dict(), stream)
 
-        return "[Mirror%s (fetch: %s, push: %s)]" % (name, self._fetch_url, self._push_url)
+    def to_dict(self):
+        return self._data
 
-    def __repr__(self):
-        return "".join(
-            (
-                "Mirror(",
-                ", ".join(
-                    "%s=%s" % (k, repr(v))
-                    for k, v in (
-                        ("fetch_url", self._fetch_url),
-                        ("push_url", self._push_url),
-                        ("name", self._name),
-                    )
-                    if k == "fetch_url" or v
-                ),
-                ")",
-            )
-        )
+    def display(self, max_len=0):
+        fetch, push = self.fetch_url, self.push_url
+        # don't print the same URL twice
+        url = fetch if fetch == push else f"fetch: {fetch} push: {push}"
+        source = "s" if self.source else " "
+        binary = "b" if self.binary else " "
+        print(f"{self.name: <{max_len}} [{source}{binary}] {url}")
 
     @property
     def name(self):
         return self._name or "<unnamed>"
 
-    def get_profile(self, url_type):
-        if isinstance(self._fetch_url, dict):
-            if url_type == "push":
-                return self._push_url.get("profile", None)
-            return self._fetch_url.get("profile", None)
-        else:
-            return None
+    @property
+    def binary(self):
+        return isinstance(self._data, str) or self._data.get("binary", True)
 
-    def set_profile(self, url_type, profile):
-        if url_type == "push":
-            self._push_url["profile"] = profile
-        else:
-            self._fetch_url["profile"] = profile
+    @property
+    def source(self):
+        return isinstance(self._data, str) or self._data.get("source", True)
 
-    def get_access_pair(self, url_type):
-        if isinstance(self._fetch_url, dict):
-            if url_type == "push":
-                return self._push_url.get("access_pair", None)
-            return self._fetch_url.get("access_pair", None)
-        else:
+    @property
+    def fetch_url(self):
+        """Get the valid, canonicalized fetch URL"""
+        return self.get_url("fetch")
+
+    @property
+    def push_url(self):
+        """Get the valid, canonicalized fetch URL"""
+        return self.get_url("push")
+
+    def _update_connection_dict(self, current_data: dict, new_data: dict, top_level: bool):
+        keys = ["url", "access_pair", "access_token", "profile", "endpoint_url"]
+        if top_level:
+            keys += ["binary", "source"]
+        changed = False
+        for key in keys:
+            if key in new_data and current_data.get(key) != new_data[key]:
+                current_data[key] = new_data[key]
+                changed = True
+        return changed
+
+    def update(self, data: dict, direction: Optional[str] = None) -> bool:
+        """Modify the mirror with the given data. This takes care
+        of expanding trivial mirror definitions by URL to something more
+        rich with a dict if necessary
+
+        Args:
+            data (dict): The data to update the mirror with.
+            direction (str): The direction to update the mirror in (fetch
+                or push or None for top-level update)
+
+        Returns:
+            bool: True if the mirror was updated, False otherwise."""
+
+        # Modify the top-level entry when no direction is given.
+        if not data:
+            return False
+
+        # If we only update a URL, there's typically no need to expand things to a dict.
+        set_url = data["url"] if len(data) == 1 and "url" in data else None
+
+        if direction is None:
+            # First deal with the case where the current top-level entry is just a string.
+            if isinstance(self._data, str):
+                # Can we replace that string with something new?
+                if set_url:
+                    if self._data == set_url:
+                        return False
+                    self._data = set_url
+                    return True
+
+                # Otherwise promote to a dict
+                self._data = {"url": self._data}
+
+            # And update the dictionary accordingly.
+            return self._update_connection_dict(self._data, data, top_level=True)
+
+        # Otherwise, update the fetch / push entry; turn top-level
+        # url string into a dict if necessary.
+        if isinstance(self._data, str):
+            self._data = {"url": self._data}
+
+        # Create a new fetch / push entry if necessary
+        if direction not in self._data:
+            # Keep config minimal if we're just setting the URL.
+            if set_url:
+                self._data[direction] = set_url
+                return True
+            self._data[direction] = {}
+
+        entry = self._data[direction]
+
+        # Keep the entry simple if we're just swapping out the URL.
+        if isinstance(entry, str):
+            if set_url:
+                if entry == set_url:
+                    return False
+                self._data[direction] = set_url
+                return True
+
+            # Otherwise promote to a dict
+            self._data[direction] = {"url": entry}
+
+        return self._update_connection_dict(self._data[direction], data, top_level=False)
+
+    def _get_value(self, attribute: str, direction: str):
+        """Returns the most specific value for a given attribute (either push/fetch or global)"""
+        if direction not in ("fetch", "push"):
+            raise ValueError(f"direction must be either 'fetch' or 'push', not {direction}")
+
+        if isinstance(self._data, str):
             return None
 
-    def set_access_pair(self, url_type, connection_tuple):
-        if url_type == "push":
-            self._push_url["access_pair"] = connection_tuple
-        else:
-            self._fetch_url["access_pair"] = connection_tuple
+        # Either a string (url) or a dictionary, we care about the dict here.
+        value = self._data.get(direction, {})
 
-    def get_endpoint_url(self, url_type):
-        if isinstance(self._fetch_url, dict):
-            if url_type == "push":
-                return self._push_url.get("endpoint_url", None)
-            return self._fetch_url.get("endpoint_url", None)
-        else:
-            return None
+        # Return top-level entry if only a URL was set.
+        if isinstance(value, str):
+            return self._data.get(attribute, None)
 
-    def set_endpoint_url(self, url_type, url):
-        if url_type == "push":
-            self._push_url["endpoint_url"] = url
-        else:
-            self._fetch_url["endpoint_url"] = url
+        return self._data.get(direction, {}).get(attribute, None)
 
-    def get_access_token(self, url_type):
-        if isinstance(self._fetch_url, dict):
-            if url_type == "push":
-                return self._push_url.get("access_token", None)
-            return self._fetch_url.get("access_token", None)
-        else:
-            return None
+    def get_url(self, direction: str):
+        if direction not in ("fetch", "push"):
+            raise ValueError(f"direction must be either 'fetch' or 'push', not {direction}")
 
-    def set_access_token(self, url_type, connection_token):
-        if url_type == "push":
-            self._push_url["access_token"] = connection_token
-        else:
-            self._fetch_url["access_token"] = connection_token
+        # Whole mirror config is just a url.
+        if isinstance(self._data, str):
+            return _url_or_path_to_url(self._data)
 
-    @property
-    def fetch_url(self):
-        """Get the valid, canonicalized fetch URL"""
-        url_or_path = (
-            self._fetch_url if isinstance(self._fetch_url, str) else self._fetch_url["url"]
-        )
-        return _url_or_path_to_url(url_or_path)
+        # Default value
+        url = self._data.get("url")
 
-    @fetch_url.setter
-    def fetch_url(self, url):
-        self._fetch_url["url"] = url
-        self._normalize()
+        # Override it with a direction-specific value
+        if direction in self._data:
+            # Either a url as string or a dict with url key
+            info = self._data[direction]
+            if isinstance(info, str):
+                url = info
+            elif "url" in info:
+                url = info["url"]
 
-    @property
-    def push_url(self):
-        """Get the valid, canonicalized push URL. Returns fetch URL if no custom
-        push URL is defined"""
-        if self._push_url is None:
-            return self.fetch_url
-        url_or_path = self._push_url if isinstance(self._push_url, str) else self._push_url["url"]
-        return _url_or_path_to_url(url_or_path)
+        return _url_or_path_to_url(url) if url else None
+
+    def get_access_token(self, direction: str):
+        return self._get_value("access_token", direction)
 
-    @push_url.setter
-    def push_url(self, url):
-        self._push_url["url"] = url
-        self._normalize()
+    def get_access_pair(self, direction: str):
+        return self._get_value("access_pair", direction)
 
-    def _normalize(self):
-        if self._push_url is not None and self._push_url == self._fetch_url:
-            self._push_url = None
+    def get_profile(self, direction: str):
+        return self._get_value("profile", direction)
+
+    def get_endpoint_url(self, direction: str):
+        return self._get_value("endpoint_url", direction)
 
 
 class MirrorCollection(collections.abc.Mapping):
     """A mapping of mirror names to mirrors."""
 
-    def __init__(self, mirrors=None, scope=None):
-        self._mirrors = collections.OrderedDict(
-            (name, Mirror.from_dict(mirror, name))
+    def __init__(
+        self,
+        mirrors=None,
+        scope=None,
+        binary: Optional[bool] = None,
+        source: Optional[bool] = None,
+    ):
+        """Initialize a mirror collection.
+
+        Args:
+            mirrors: A name-to-mirror mapping to initialize the collection with.
+            scope: The scope to use when looking up mirrors from the config.
+            binary: If True, only include binary mirrors.
+                    If False, omit binary mirrors.
+                    If None, do not filter on binary mirrors.
+            source: If True, only include source mirrors.
+                    If False, omit source mirrors.
+                    If None, do not filter on source mirrors."""
+        self._mirrors = {
+            name: Mirror(data=mirror, name=name)
             for name, mirror in (
                 mirrors.items()
                 if mirrors is not None
                 else spack.config.get("mirrors", scope=scope).items()
             )
-        )
+        }
+
+        if source is not None:
+            self._mirrors = {k: v for k, v in self._mirrors.items() if v.source == source}
+
+        if binary is not None:
+            self._mirrors = {k: v for k, v in self._mirrors.items() if v.binary == binary}
 
     def __eq__(self, other):
         return self._mirrors == other._mirrors
@@ -325,7 +360,7 @@ class MirrorCollection(collections.abc.Mapping):
         result = self.get(name_or_url)
 
         if result is None:
-            result = Mirror(fetch_url=name_or_url)
+            result = Mirror(fetch=name_or_url)
 
         return result
 
@@ -576,24 +611,8 @@ def remove(name, scope):
     if name not in mirrors:
         tty.die("No mirror with name %s" % name)
 
-    old_value = mirrors.pop(name)
+    mirrors.pop(name)
     spack.config.set("mirrors", mirrors, scope=scope)
-
-    debug_msg_url = "url %s"
-    debug_msg = ["Removed mirror %s with"]
-    values = [name]
-
-    try:
-        fetch_value = old_value["fetch"]
-        push_value = old_value["push"]
-
-        debug_msg.extend(("fetch", debug_msg_url, "and push", debug_msg_url))
-        values.extend((fetch_value, push_value))
-    except TypeError:
-        debug_msg.append(debug_msg_url)
-        values.append(old_value)
-
-    tty.debug(" ".join(debug_msg) % tuple(values))
     tty.msg("Removed mirror %s." % name)
 
 
diff --git a/lib/spack/spack/schema/mirrors.py b/lib/spack/spack/schema/mirrors.py
index 41dbddb051..8001172afd 100644
--- a/lib/spack/spack/schema/mirrors.py
+++ b/lib/spack/spack/schema/mirrors.py
@@ -6,29 +6,55 @@
 """Schema for mirrors.yaml configuration file.
 
 .. literalinclude:: _spack_root/lib/spack/spack/schema/mirrors.py
+   :lines: 12-69
 """
 
+#: Common properties for connection specification
+connection = {
+    "url": {"type": "string"},
+    # todo: replace this with named keys "username" / "password" or "id" / "secret"
+    "access_pair": {
+        "type": "array",
+        "items": {"type": ["string", "null"], "minItems": 2, "maxItems": 2},
+    },
+    "access_token": {"type": ["string", "null"]},
+    "profile": {"type": ["string", "null"]},
+    "endpoint_url": {"type": ["string", "null"]},
+}
+
+#: Mirror connection inside pull/push keys
+fetch_and_push = {
+    "anyOf": [
+        {"type": "string"},
+        {
+            "type": "object",
+            "additionalProperties": False,
+            "properties": {**connection},  # type: ignore
+        },
+    ]
+}
+
+#: Mirror connection when no pull/push keys are set
+mirror_entry = {
+    "type": "object",
+    "additionalProperties": False,
+    "anyOf": [{"required": ["url"]}, {"required": ["fetch"]}, {"required": ["pull"]}],
+    "properties": {
+        "source": {"type": "boolean"},
+        "binary": {"type": "boolean"},
+        "fetch": fetch_and_push,
+        "push": fetch_and_push,
+        **connection,  # type: ignore
+    },
+}
+
 #: Properties for inclusion in other schemas
 properties = {
     "mirrors": {
         "type": "object",
         "default": {},
         "additionalProperties": False,
-        "patternProperties": {
-            r"\w[\w-]*": {
-                "anyOf": [
-                    {"type": "string"},
-                    {
-                        "type": "object",
-                        "required": ["fetch", "push"],
-                        "properties": {
-                            "fetch": {"type": ["string", "object"]},
-                            "push": {"type": ["string", "object"]},
-                        },
-                    },
-                ]
-            }
-        },
+        "patternProperties": {r"\w[\w-]*": {"anyOf": [{"type": "string"}, mirror_entry]}},
     }
 }
 
diff --git a/lib/spack/spack/stage.py b/lib/spack/spack/stage.py
index 09ea1f88a4..2dd3a98a55 100644
--- a/lib/spack/spack/stage.py
+++ b/lib/spack/spack/stage.py
@@ -449,22 +449,11 @@ class Stage:
             # Join URLs of mirror roots with mirror paths. Because
             # urljoin() will strip everything past the final '/' in
             # the root, so we add a '/' if it is not present.
-            mirror_urls = {}
-            for mirror in spack.mirror.MirrorCollection().values():
-                for rel_path in self.mirror_paths:
-                    mirror_url = url_util.join(mirror.fetch_url, rel_path)
-                    mirror_urls[mirror_url] = {}
-                    if (
-                        mirror.get_access_pair("fetch")
-                        or mirror.get_access_token("fetch")
-                        or mirror.get_profile("fetch")
-                    ):
-                        mirror_urls[mirror_url] = {
-                            "access_token": mirror.get_access_token("fetch"),
-                            "access_pair": mirror.get_access_pair("fetch"),
-                            "access_profile": mirror.get_profile("fetch"),
-                            "endpoint_url": mirror.get_endpoint_url("fetch"),
-                        }
+            mirror_urls = [
+                url_util.join(mirror.fetch_url, rel_path)
+                for mirror in spack.mirror.MirrorCollection(source=True).values()
+                for rel_path in self.mirror_paths
+            ]
 
             # If this archive is normally fetched from a tarball URL,
             # then use the same digest.  `spack mirror` ensures that
@@ -483,16 +472,9 @@ class Stage:
 
             # Add URL strategies for all the mirrors with the digest
             # Insert fetchers in the order that the URLs are provided.
-            for url in reversed(list(mirror_urls.keys())):
+            for url in reversed(mirror_urls):
                 fetchers.insert(
-                    0,
-                    fs.from_url_scheme(
-                        url,
-                        digest,
-                        expand=expand,
-                        extension=extension,
-                        connection=mirror_urls[url],
-                    ),
+                    0, fs.from_url_scheme(url, digest, expand=expand, extension=extension)
                 )
 
             if self.default_fetcher.cachable:
diff --git a/lib/spack/spack/test/cmd/mirror.py b/lib/spack/spack/test/cmd/mirror.py
index 6bb3ae218f..0b9697976f 100644
--- a/lib/spack/spack/test/cmd/mirror.py
+++ b/lib/spack/spack/test/cmd/mirror.py
@@ -149,7 +149,7 @@ def test_mirror_crud(mutable_config, capsys):
         assert "No changes made" in output
 
         output = mirror("set-url", "--push", "mirror", "s3://spack-public")
-        assert "Changed (push) url" in output
+        assert not output
 
         # no-op
         output = mirror("set-url", "--push", "mirror", "s3://spack-public")
@@ -348,3 +348,54 @@ class TestMirrorCreate:
         args = MockMirrorArgs(specs=input_specs, versions_per_spec=nversions)
         specs = spack.cmd.mirror.concrete_specs_from_user(args)
         assert all(s.concrete for s in specs)
+
+
+def test_mirror_type(mutable_config):
+    """Test the mirror set command"""
+    mirror("add", "example", "--type", "binary", "http://example.com")
+    assert spack.config.get("mirrors:example") == {
+        "url": "http://example.com",
+        "source": False,
+        "binary": True,
+    }
+
+    mirror("set", "example", "--type", "source")
+    assert spack.config.get("mirrors:example") == {
+        "url": "http://example.com",
+        "source": True,
+        "binary": False,
+    }
+
+    mirror("set", "example", "--type", "binary")
+    assert spack.config.get("mirrors:example") == {
+        "url": "http://example.com",
+        "source": False,
+        "binary": True,
+    }
+    mirror("set", "example", "--type", "binary", "--type", "source")
+    assert spack.config.get("mirrors:example") == {
+        "url": "http://example.com",
+        "source": True,
+        "binary": True,
+    }
+
+
+def test_mirror_set_2(mutable_config):
+    """Test the mirror set command"""
+    mirror("add", "example", "http://example.com")
+    mirror(
+        "set",
+        "example",
+        "--push",
+        "--url",
+        "http://example2.com",
+        "--s3-access-key-id",
+        "username",
+        "--s3-access-key-secret",
+        "password",
+    )
+
+    assert spack.config.get("mirrors:example") == {
+        "url": "http://example.com",
+        "push": {"url": "http://example2.com", "access_pair": ["username", "password"]},
+    }
diff --git a/lib/spack/spack/test/mirror.py b/lib/spack/spack/test/mirror.py
index 5d5970c2af..0ed89322c1 100644
--- a/lib/spack/spack/test/mirror.py
+++ b/lib/spack/spack/test/mirror.py
@@ -132,9 +132,14 @@ def test_all_mirror(mock_git_repository, mock_svn_repository, mock_hg_repository
 
 
 @pytest.mark.parametrize(
-    "mirror", [spack.mirror.Mirror("https://example.com/fetch", "https://example.com/push")]
+    "mirror",
+    [
+        spack.mirror.Mirror(
+            {"fetch": "https://example.com/fetch", "push": "https://example.com/push"}
+        )
+    ],
 )
-def test_roundtrip_mirror(mirror):
+def test_roundtrip_mirror(mirror: spack.mirror.Mirror):
     mirror_yaml = mirror.to_yaml()
     assert spack.mirror.Mirror.from_yaml(mirror_yaml) == mirror
     mirror_json = mirror.to_json()
@@ -291,3 +296,70 @@ def test_get_all_versions(specs, expected_specs):
     output_list = [str(x) for x in output_list]
     # Compare sets since order is not important
     assert set(output_list) == set(expected_specs)
+
+
+def test_update_1():
+    # No change
+    m = spack.mirror.Mirror("https://example.com")
+    assert not m.update({"url": "https://example.com"})
+    assert m.to_dict() == "https://example.com"
+
+
+def test_update_2():
+    # Change URL, shouldn't expand to {"url": ...} dict.
+    m = spack.mirror.Mirror("https://example.com")
+    assert m.update({"url": "https://example.org"})
+    assert m.to_dict() == "https://example.org"
+    assert m.fetch_url == "https://example.org"
+    assert m.push_url == "https://example.org"
+
+
+def test_update_3():
+    # Change fetch url, ensure minimal config
+    m = spack.mirror.Mirror("https://example.com")
+    assert m.update({"url": "https://example.org"}, "fetch")
+    assert m.to_dict() == {"url": "https://example.com", "fetch": "https://example.org"}
+    assert m.fetch_url == "https://example.org"
+    assert m.push_url == "https://example.com"
+
+
+def test_update_4():
+    # Change push url, ensure minimal config
+    m = spack.mirror.Mirror("https://example.com")
+    assert m.update({"url": "https://example.org"}, "push")
+    assert m.to_dict() == {"url": "https://example.com", "push": "https://example.org"}
+    assert m.push_url == "https://example.org"
+    assert m.fetch_url == "https://example.com"
+
+
+@pytest.mark.parametrize("direction", ["fetch", "push"])
+def test_update_connection_params(direction):
+    """Test whether new connection params expand the mirror config to a dict."""
+    m = spack.mirror.Mirror("https://example.com")
+
+    assert m.update(
+        {
+            "url": "http://example.org",
+            "access_pair": ["username", "password"],
+            "access_token": "token",
+            "profile": "profile",
+            "endpoint_url": "https://example.com",
+        },
+        direction,
+    )
+
+    assert m.to_dict() == {
+        "url": "https://example.com",
+        direction: {
+            "url": "http://example.org",
+            "access_pair": ["username", "password"],
+            "access_token": "token",
+            "profile": "profile",
+            "endpoint_url": "https://example.com",
+        },
+    }
+
+    assert m.get_access_pair(direction) == ["username", "password"]
+    assert m.get_access_token(direction) == "token"
+    assert m.get_profile(direction) == "profile"
+    assert m.get_endpoint_url(direction) == "https://example.com"
diff --git a/lib/spack/spack/test/web.py b/lib/spack/spack/test/web.py
index b681b5b0a0..6d451ed20a 100644
--- a/lib/spack/spack/test/web.py
+++ b/lib/spack/spack/test/web.py
@@ -267,7 +267,7 @@ class MockS3Client:
 
 
 def test_gather_s3_information(monkeypatch, capfd):
-    mirror = spack.mirror.Mirror.from_dict(
+    mirror = spack.mirror.Mirror(
         {
             "fetch": {
                 "access_token": "AAAAAAA",
diff --git a/share/spack/spack-completion.bash b/share/spack/spack-completion.bash
index 959e4924f2..e9e28dcfd9 100755
--- a/share/spack/spack-completion.bash
+++ b/share/spack/spack-completion.bash
@@ -1290,7 +1290,7 @@ _spack_mirror() {
     then
         SPACK_COMPREPLY="-h --help -n --no-checksum --deprecated"
     else
-        SPACK_COMPREPLY="create destroy add remove rm set-url list"
+        SPACK_COMPREPLY="create destroy add remove rm set-url set list"
     fi
 }
 
@@ -1310,7 +1310,7 @@ _spack_mirror_destroy() {
 _spack_mirror_add() {
     if $list_options
     then
-        SPACK_COMPREPLY="-h --help --scope --s3-access-key-id --s3-access-key-secret --s3-access-token --s3-profile --s3-endpoint-url"
+        SPACK_COMPREPLY="-h --help --scope --type --s3-access-key-id --s3-access-key-secret --s3-access-token --s3-profile --s3-endpoint-url"
     else
         _mirrors
     fi
@@ -1337,7 +1337,16 @@ _spack_mirror_rm() {
 _spack_mirror_set_url() {
     if $list_options
     then
-        SPACK_COMPREPLY="-h --help --push --scope --s3-access-key-id --s3-access-key-secret --s3-access-token --s3-profile --s3-endpoint-url"
+        SPACK_COMPREPLY="-h --help --push --fetch --scope --s3-access-key-id --s3-access-key-secret --s3-access-token --s3-profile --s3-endpoint-url"
+    else
+        _mirrors
+    fi
+}
+
+_spack_mirror_set() {
+    if $list_options
+    then
+        SPACK_COMPREPLY="-h --help --push --fetch --type --url --scope --s3-access-key-id --s3-access-key-secret --s3-access-token --s3-profile --s3-endpoint-url"
     else
         _mirrors
     fi