diff --git a/charms/worker/charmcraft.yaml b/charms/worker/charmcraft.yaml
index ba4e94f3..96a5f9ca 100644
--- a/charms/worker/charmcraft.yaml
+++ b/charms/worker/charmcraft.yaml
@@ -82,3 +82,5 @@ requires:
     #   juju integrate k8s:k8s-cluster k8s-worker:cluster
   cos-tokens:
     interface: cos-k8s-tokens
+  containerd:
+    interface: containerd
diff --git a/charms/worker/k8s/charmcraft.yaml b/charms/worker/k8s/charmcraft.yaml
index 7dc97146..bcec2458 100644
--- a/charms/worker/k8s/charmcraft.yaml
+++ b/charms/worker/k8s/charmcraft.yaml
@@ -54,6 +54,53 @@ bases:
       architectures: [amd64]
 config:
   options:
+    containerd_custom_registries:
+      type: string
+      default: "[]"
+      description: |
+        Registry endpoints and credentials. Setting this config allows containerd
+        to pull images from registries where auth is required.
+
+        The value for this config must be a JSON array of credential objects, like this:
+          e.g.: [{"url": "https://registry.example.com", "host": "my.registry:port", "username": "user", "password": "pass"}]
+
+        Credential Object Parameters:
+        url: REQUIRED str
+          the URL to the registry, include the port if not it isn't implied from the schema.
+            e.g: "url": "https://my.registry:8443"
+            e.g: "url": "http://my.registry"
+
+        host: OPTIONAL str - defaults to auto-generated from the url
+          could be registry host address or a name
+            e.g.: myregistry.io:9000, 10.10.10.10:5432
+            e.g.: myregistry.io, myregistry
+          Note: It will be derived from `url` if not provided.
+            e.g.: "url": "http://10.10.10.10:8000" --> "host": "10.10.10.10:8000"
+
+        username: OPTIONAL str - default ''
+        password: OPTIONAL str - default ''
+        identitytoken: OPTIONAL str - default ''
+          Used by containerd for basic authentication to the registry.
+
+        ca_file: OPTIONAL str - default ''
+        cert_file: OPTIONAL str - default ''
+        key_file: OPTIONAL str - default ''
+          For ssl/tls communication these should be a base64 encoded file
+          e.g.:  "ca_file": "'"$(base64 -w 0 < my.custom.registry.pem)"'"
+
+        skip_verify: OPTIONAL bool - default false
+          For situations where the registry has self-signed or expired certs and a quick work-around is necessary.
+          e.g.: "skip_verify": true
+
+        example config)
+        juju config containerd custom_registries='[{
+            "url": "https://registry.example.com",
+            "host": "ghcr.io",
+            "ca_file": "'"$(base64 -w 0 < ~/my.custom.ca.pem)"'",
+            "cert_file": "'"$(base64 -w 0 < ~/my.custom.cert.pem)"'",
+            "key_file": "'"$(base64 -w 0 < ~/my.custom.key.pem)"'",
+        }]'
+
     datastore:
       default: dqlite
       type: string
@@ -113,6 +160,8 @@ provides:
     interface: k8s-cluster
   cos-worker-tokens:
     interface: cos-k8s-tokens
+  containerd:
+    interface: containerd
 
 requires:
   etcd:
diff --git a/charms/worker/k8s/requirements.txt b/charms/worker/k8s/requirements.txt
index 399a6c7b..d3ccfc26 100644
--- a/charms/worker/k8s/requirements.txt
+++ b/charms/worker/k8s/requirements.txt
@@ -6,5 +6,7 @@ cosl==0.0.8
 ops==2.14.0
 pydantic==1.10.15
 PyYAML==6.0.1
+tomli == 2.0.1
+tomli-w == 1.0.0
 typing_extensions==4.12.0
 websocket-client==1.8.0
diff --git a/charms/worker/k8s/src/charm.py b/charms/worker/k8s/src/charm.py
index ddaceb39..0d042950 100755
--- a/charms/worker/k8s/src/charm.py
+++ b/charms/worker/k8s/src/charm.py
@@ -29,6 +29,7 @@
 
 import charms.contextual_status as status
 import charms.operator_libs_linux.v2.snap as snap_lib
+import containerd
 import ops
 import reschedule
 import yaml
@@ -297,6 +298,26 @@ def _bootstrap_k8s_snap(self):
         # TODO: Make port (and address) configurable.
         self.api_manager.bootstrap_k8s_snap(payload)
 
+    @on_error(
+        ops.BlockedStatus("Failed to apply containerd_custom_registries, check logs for details"),
+        ValueError,
+        subprocess.CalledProcessError,
+        OSError,
+    )
+    def _config_containerd_registries(self):
+        """Apply containerd custom registries."""
+        registries, config = [], ""
+        containerd_relation = self.model.get_relation("containerd")
+        if self.is_control_plane:
+            config = self.config["containerd_custom_registries"]
+            registries = containerd.parse_registries(config)
+        else:
+            registries = containerd.recover(containerd_relation)
+        self.unit.status = ops.MaintenanceStatus("Ensuring containerd registries")
+        containerd.ensure_registry_configs(registries)
+        if self.lead_control_plane:
+            containerd.share(config, self.app, containerd_relation)
+
     def _configure_cos_integration(self):
         """Retrieve the join token from secret databag and join the cluster."""
         if not self.model.get_relation("cos-agent"):
@@ -588,6 +609,7 @@ def _reconcile(self, event: ops.EventBase):
             self._revoke_cluster_tokens(event)
             self._ensure_cluster_config()
         self._join_cluster()
+        self._config_containerd_registries()
         self._configure_cos_integration()
         self._update_status()
         self._apply_node_labels()
diff --git a/charms/worker/k8s/src/containerd.py b/charms/worker/k8s/src/containerd.py
new file mode 100644
index 00000000..3b5a3908
--- /dev/null
+++ b/charms/worker/k8s/src/containerd.py
@@ -0,0 +1,334 @@
+# Copyright 2024 Canonical Ltd.
+# See LICENSE file for licensing details.
+
+# Learn more at: https://juju.is/docs/sdk
+
+"""Configuration for containerd.
+
+The format for the hosts.toml file is as follows:
+https://github.com/containerd/containerd/blob/main/docs/hosts.md
+
+The format for the config.toml file is as follows:
+https://github.com/containerd/containerd/blob/main/docs/cri/registry.md
+"""
+
+import base64
+import collections
+import json
+import logging
+import os
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+from urllib.parse import urlparse
+
+import ops
+import pydantic
+import tomli_w
+
+log = logging.getLogger(__name__)
+HOSTSD_PATH = Path("/var/snap/k8s/common/etc/containerd/hosts.d/")
+
+
+def _ensure_file(
+    file: Path,
+    data: str,
+    permissions: Optional[int] = None,
+    uid: Optional[int] = None,
+    gid: Optional[int] = None,
+) -> bool:
+    """Ensure file with specific contents, owner:group and permissions exists on disk.
+
+    Args:
+        file (Path):       path to the file
+        data (str):        content of the file
+        permissions (int): permissions on the file
+        uid (int):         user owner id
+        gid (int):         group owner id
+
+    Returns:
+        `True` - if file contents have changed
+    """
+    file.parent.mkdir(parents=True, exist_ok=True)
+
+    changed = False
+    if not file.exists() or file.read_text() != data:
+        file.write_text(data)
+        changed = True
+
+    if permissions is not None:
+        file.chmod(permissions)
+
+    if uid is not None and gid is not None:
+        os.chown(file, uid, gid)
+
+    return changed
+
+
+class Registry(pydantic.BaseModel, extra=pydantic.Extra.forbid):
+    """Represents a containerd registry.
+
+    Attrs:
+        url (HttpUrl):
+        host (str):
+        username (SecretStr):
+        password (SecretStr):
+        identitytoken (SecretStr):
+        ca_file (str):
+        cert_file (str):
+        key_file (str):
+        skip_verify (bool):
+        override_path (bool):
+        ca_file_path (Path):
+        cert_file_path (Path):
+        key_file_path (Path):
+        hosts_toml_path (Path):
+        auth_config_header (Dict[str, Any]):
+        hosts_toml (Dict[str, Any]):
+    """
+
+    # e.g. "https://registry-1.docker.io"
+    url: pydantic.AnyHttpUrl
+
+    # e.g. "docker.io", or "registry.example.com:32000"
+    host: str = ""
+
+    # authentication settings
+    username: Optional[pydantic.SecretStr] = None
+    password: Optional[pydantic.SecretStr] = None
+    identitytoken: Optional[pydantic.SecretStr] = None
+
+    # TLS configuration
+    ca_file: Optional[str] = None
+    cert_file: Optional[str] = None
+    key_file: Optional[str] = None
+    skip_verify: Optional[bool] = None
+
+    # misc configuration
+    override_path: Optional[bool] = None
+
+    def __init__(self, *args, **kwargs):
+        """Create a registry object.
+
+        Args:
+            args:   construction positional arguments
+            kwargs: construction keyword arguments
+        """
+        super(Registry, self).__init__(*args, **kwargs)
+        if not self.host and (host := urlparse(self.url).netloc):
+            self.host = host
+
+    @pydantic.validator("ca_file", "cert_file", "key_file")
+    def parse_base64(cls, v: str) -> str:
+        """Validate Base64 Content.
+
+        Args:
+            v (str): value to validate
+
+        Returns:
+            validated content
+        """
+        return base64.b64decode(v.encode()).decode()
+
+    @property
+    def ca_file_path(self) -> Path:
+        """Return CA file path.
+
+        Returns:
+            path to file
+        """
+        return HOSTSD_PATH / self.host / "ca.crt"
+
+    @property
+    def cert_file_path(self) -> Path:
+        """Return Cert file path.
+
+        Returns:
+            path to file
+        """
+        return HOSTSD_PATH / self.host / "client.crt"
+
+    @property
+    def key_file_path(self) -> Path:
+        """Return Key file path.
+
+        Returns:
+            path to file
+        """
+        return HOSTSD_PATH / self.host / "client.key"
+
+    @property
+    def hosts_toml_path(self) -> Path:
+        """Return hosts.toml path.
+
+        Returns:
+            path to file
+        """
+        return HOSTSD_PATH / self.host / "hosts.toml"
+
+    @property
+    def auth_config_header(self) -> Dict[str, Any]:
+        """Return a fixed auth configuration header for registry.
+
+        TODO: May need to be extended for other auth methods (eg. oauth2, etc.)
+
+        Returns:
+            This registry's auth content headers
+        """
+        if self.username and self.password:
+            log.debug("Configure basic auth for %s (%s)", self.url, self.host)
+            v = self.username.get_secret_value() + ":" + self.password.get_secret_value()
+            return {"Authorization": "Basic " + base64.b64encode(v.encode()).decode()}
+        elif self.identitytoken:
+            log.debug("Configure bearer token for %s (%s)", self.url, self.host)
+            return {"Authorization": "Bearer " + self.identitytoken.get_secret_value()}
+        else:
+            return {}
+
+    @property
+    def hosts_toml(self) -> Dict[str, Any]:
+        """Return data for hosts.toml file.
+
+        Returns:
+            hosts.toml content
+        """
+        host_config: Dict[str, Any] = {"capabilities": ["pull", "resolve"]}
+        if self.ca_file:
+            host_config["ca"] = self.ca_file_path.as_posix()
+        if self.cert_file and self.key_file:
+            host_config["client"] = [
+                [self.cert_file_path.as_posix(), self.key_file_path.as_posix()]
+            ]
+        elif self.cert_file:
+            host_config["client"] = self.cert_file_path.as_posix()
+
+        if self.skip_verify:
+            host_config["skip_verify"] = True
+        if self.override_path:
+            host_config["override_path"] = True
+        if config := self.auth_config_header:
+            host_config["header"] = config
+
+        return {
+            "server": self.url,
+            "host": {self.url: host_config},
+        }
+
+    def ensure_certificates(self):
+        """Ensure client and ca certificates."""
+        ca_file_path = self.ca_file_path
+        if self.ca_file:
+            log.debug("Configure custom CA path %s", ca_file_path)
+            _ensure_file(ca_file_path, self.ca_file, 0o600, 0, 0)
+        else:
+            ca_file_path.unlink(missing_ok=True)
+
+        cert_file_path = self.cert_file_path
+        if self.cert_file:
+            log.debug("Configure client certificate path %s", cert_file_path)
+            _ensure_file(cert_file_path, self.cert_file, 0o600, 0, 0)
+        else:
+            cert_file_path.unlink(missing_ok=True)
+
+        key_file_path = self.key_file_path
+        if self.key_file:
+            log.debug("Configure client key path %s", key_file_path)
+            _ensure_file(key_file_path, self.key_file, 0o600, 0, 0)
+        else:
+            key_file_path.unlink(missing_ok=True)
+
+    def ensure_hosts_toml(self):
+        """Ensure hosts.toml file."""
+        hosts_toml_path = self.hosts_toml_path
+        log.debug("Configure hosts.toml %s", hosts_toml_path)
+        _ensure_file(hosts_toml_path, tomli_w.dumps(self.hosts_toml), 0o600, 0, 0)
+
+
+class RegistryConfigs(pydantic.BaseModel, extra=pydantic.Extra.forbid):
+    """Represents a set of containerd registries.
+
+    Attrs:
+        registries (List[Registry]):
+    """
+
+    registries: List[Registry]
+
+
+def parse_registries(json_str: str) -> List[Registry]:
+    """Parse registry configurations from json string.
+
+    Args:
+        json_str (str): raw user supplied content
+
+    Returns:
+        RegistryConfigs parsed from json_str
+
+    Raises:
+        ValueError: if configuration is not valid
+    """
+    if not json_str:
+        return []
+
+    try:
+        parsed = json.loads(json_str)
+    except json.JSONDecodeError as e:
+        raise ValueError(f"not valid JSON: {e}") from e
+
+    parsed = RegistryConfigs(registries=parsed)
+    dupes = [x for x, y in collections.Counter(x.host for x in parsed.registries).items() if y > 1]
+    if len(dupes):
+        raise ValueError(f"duplicate host definitions: {','.join(dupes)}")
+    return parsed.registries
+
+
+def ensure_registry_configs(registries: List[Registry]):
+    """Ensure containerd configuration files match the specified registries.
+
+    Args:
+        registries (List[Registry]): list of registries
+    """
+    unneeded = {host.parent.name for host in HOSTSD_PATH.glob("**/hosts.toml")}
+    for r in registries:
+        unneeded -= {r.host}
+        log.info("Configure registry %s (%s)", r.host, r.url)
+        r.ensure_certificates()
+        r.ensure_hosts_toml()
+
+    for h in unneeded:
+        log.info("Removing unneeded registry %s", h)
+        (HOSTSD_PATH / h / "hosts.toml").unlink(missing_ok=True)
+
+
+def share(config: str, app: ops.Application, relation: Optional[ops.Relation]):
+    """Share containerd configuration over relation application databag.
+
+    Args:
+        config (str): list of registries
+        app (ops.Application): application to share with.
+        relation (ops.Relation): relation on which to share.
+    """
+    if not relation:
+        log.info("No relation to share containerd config.")
+        return
+    relation.data[app]["custom-registries"] = config
+
+
+def recover(relation: Optional[ops.Relation]) -> List[Registry]:
+    """Share containerd configuration over relation application databag.
+
+    Args:
+        relation (ops.Relation): relation on which to receive.
+
+    Returns:
+        RegistryConfigs parsed from json_str
+    """
+    if not relation:
+        log.info("No relation to recover containerd config.")
+        return []
+    if not (app_databag := relation.data.get(relation.app)):
+        log.warning("No application data to recover containerd config.")
+        return []
+    if not (config := app_databag.get("custom-registries")):
+        log.warning("No 'custom-registries' to recover containerd config.")
+        return []
+    log.info("Recovering containerd from relation %s", relation.id)
+    return parse_registries(config)
diff --git a/charms/worker/k8s/tests/unit/test_containerd.py b/charms/worker/k8s/tests/unit/test_containerd.py
new file mode 100644
index 00000000..6e00ade3
--- /dev/null
+++ b/charms/worker/k8s/tests/unit/test_containerd.py
@@ -0,0 +1,188 @@
+# Copyright 2024 Canonical Ltd.
+# See LICENSE file for licensing details.
+
+# Learn more about testing at: https://juju.is/docs/sdk/testing
+
+"""Unit tests containerd module."""
+import unittest.mock as mock
+from os import getgid, getuid
+
+import containerd
+import pytest
+import tomli_w
+
+
+def test_ensure_file(tmp_path):
+    """Test ensure file method."""
+    test_file = tmp_path / "test.txt"
+    assert containerd._ensure_file(test_file, "data", 0o644, getuid(), getgid())
+    assert test_file.read_text() == "data"
+    assert test_file.stat().st_mode == 0o100644
+    assert test_file.stat().st_uid == getuid()
+    assert test_file.stat().st_gid == getgid()
+
+
+def test_registry_parse_default():
+    """Test default registry parsing."""
+    assert containerd.parse_registries("[]") == []
+
+
+def test_registry_parse_all_fields():
+    """Test default registry parsing all fields."""
+    parsed = containerd.parse_registries(
+        """[{
+        "host": "ghcr.io",
+        "url": "http://ghcr.io",
+        "ca_file": "Y2FfZmlsZQ==",
+        "cert_file": "Y2VydF9maWxl",
+        "key_file": "a2V5X2ZpbGU=",
+        "username": "user",
+        "password": "pass",
+        "identitytoken": "token",
+        "skip_verify": true,
+        "override_path": true
+    }]"""
+    )
+    expected = containerd.Registry(
+        host="ghcr.io",
+        url="http://ghcr.io",
+        ca_file="Y2FfZmlsZQ==",
+        cert_file="Y2VydF9maWxl",
+        key_file="a2V5X2ZpbGU=",
+        username="user",
+        password="pass",
+        identitytoken="token",
+        skip_verify=True,
+        override_path=True,
+    )
+    assert parsed == [expected]
+
+
+@pytest.mark.parametrize(
+    "registry_errors",
+    [
+        ("{", "not valid JSON"),
+        ("{}", "value is not a valid list"),
+        ("[1]", "value is not a valid dict"),
+        ("[{}]", "url\n  field required"),
+        ('[{"url": 1}]', "invalid or missing URL scheme"),
+        (
+            '[{"url": "http://ghcr.io", "why-am-i-here": "abc"}]',
+            "extra fields not permitted",
+        ),
+        (
+            '[{"url": "http://ghcr.io"}, {"url": "http://ghcr.io"}]',
+            "duplicate host definitions: ghcr.io",
+        ),
+    ],
+    ids=[
+        "Invalid JSON",
+        "Not a List",
+        "List Item not an object",
+        "Missing required field",
+        "Invalid URL",
+        "Restricted field",
+        "Duplicate host",
+    ],
+)
+def test_registry_parse_failures(registry_errors):
+    """Test default registry parsing."""
+    registries, expected = registry_errors
+    with pytest.raises(ValueError) as e:
+        containerd.parse_registries(registries)
+    assert expected in str(e.value)
+
+
+def test_registry_methods():
+    """Test registry methods."""
+    registry = containerd.Registry(
+        host="ghcr-mirror.io",
+        url="http://ghcr.io",
+        ca_file="Y2FfZmlsZQ==",
+        cert_file="Y2VydF9maWxl",
+        key_file="a2V5X2ZpbGU=",
+        username="user",
+        password="pass",
+        identitytoken="token",
+        skip_verify=True,
+        override_path=True,
+    )
+
+    assert registry.ca_file_path == containerd.HOSTSD_PATH / "ghcr-mirror.io/ca.crt"
+    assert registry.cert_file_path == containerd.HOSTSD_PATH / "ghcr-mirror.io/client.crt"
+    assert registry.key_file_path == containerd.HOSTSD_PATH / "ghcr-mirror.io/client.key"
+    assert registry.hosts_toml_path == containerd.HOSTSD_PATH / "ghcr-mirror.io/hosts.toml"
+
+    assert registry.auth_config_header == {"Authorization": "Basic dXNlcjpwYXNz"}
+
+    registry.password = None
+    assert registry.auth_config_header == {"Authorization": "Bearer token"}
+
+    assert registry.hosts_toml == {
+        "server": "http://ghcr.io",
+        "host": {
+            "http://ghcr.io": {
+                "capabilities": ["pull", "resolve"],
+                "ca": str(registry.ca_file_path),
+                "client": [[str(registry.cert_file_path), str(registry.key_file_path)]],
+                "skip_verify": True,
+                "override_path": True,
+                "header": {"Authorization": "Bearer token"},
+            },
+        },
+    }
+    registry.key_file = None
+    assert registry.hosts_toml == {
+        "server": "http://ghcr.io",
+        "host": {
+            "http://ghcr.io": {
+                "capabilities": ["pull", "resolve"],
+                "ca": str(registry.ca_file_path),
+                "client": str(registry.cert_file_path),
+                "skip_verify": True,
+                "override_path": True,
+                "header": {"Authorization": "Bearer token"},
+            },
+        },
+    }
+
+    registry.key_file = "key_file"
+    with mock.patch("containerd._ensure_file") as ensure_file:
+        registry.ensure_certificates()
+        ensure_file.assert_has_calls(
+            [
+                mock.call(registry.ca_file_path, "ca_file", 0o600, 0, 0),
+                mock.call(registry.cert_file_path, "cert_file", 0o600, 0, 0),
+                mock.call(registry.key_file_path, "key_file", 0o600, 0, 0),
+            ]
+        )
+
+    with mock.patch("containerd._ensure_file") as ensure_file:
+        registry.ensure_hosts_toml()
+        ensure_file.assert_has_calls(
+            [
+                mock.call(
+                    registry.hosts_toml_path, tomli_w.dumps(registry.hosts_toml), 0o600, 0, 0
+                ),
+            ]
+        )
+
+
+@mock.patch("containerd._ensure_file")
+def test_ensure_registry_configs(mock_ensure_file):
+    """Test registry methods."""
+    registry = containerd.Registry(
+        host="ghcr-mirror.io",
+        url="http://ghcr.io",
+        ca_file="Y2FfZmlsZQ==",
+        cert_file="Y2VydF9maWxl",
+        key_file="a2V5X2ZpbGU=",
+        username="user",
+        password="pass",
+        identitytoken="token",
+        skip_verify=True,
+        override_path=True,
+    )
+
+    containerd.ensure_registry_configs([registry])
+    assert mock_ensure_file.call_count == 4, "4 files should be written"
diff --git a/charms/worker/k8s/tox.ini b/charms/worker/k8s/tox.ini
index 9d9d5d5d..1d95f71e 100644
--- a/charms/worker/k8s/tox.ini
+++ b/charms/worker/k8s/tox.ini
@@ -31,7 +31,9 @@ deps =
     -r{toxinidir}/requirements.txt
 commands =
     coverage run --source={[vars]src_path},{[vars]lib_path} \
-        -m pytest --ignore={[vars]tst_path}integration -v --tb native -s {posargs}
+        -m pytest --ignore={[vars]tst_path}integration -vv \
+        --basetemp={envtmpdir} \
+        --tb native -s {posargs}
     coverage report --show-missing
 
 [testenv:coverage-report]
diff --git a/tox.ini b/tox.ini
index 4d33fb87..0e1e1946 100644
--- a/tox.ini
+++ b/tox.ini
@@ -67,7 +67,7 @@ commands =
 [testenv:unit]
 allowlist_externals = tox
 commands =
-    tox -c {toxinidir}/charms/worker/k8s -e unit
+    tox -c {toxinidir}/charms/worker/k8s -e unit -- {posargs}
 
 [testenv:coverage-report]
 allowlist_externals = tox