diff --git a/CODEOWNERS b/CODEOWNERS
index 57f2174a35f9..3037314ccdbb 100644
--- a/CODEOWNERS
+++ b/CODEOWNERS
@@ -1107,6 +1107,8 @@ CLAUDE.md @home-assistant/core
 /tests/components/lyric/ @timmo001
 /homeassistant/components/madvr/ @iloveicedgreentea
 /tests/components/madvr/ @iloveicedgreentea
+/homeassistant/components/map_tiles/ @home-assistant/core
+/tests/components/map_tiles/ @home-assistant/core
 /homeassistant/components/marantz_infrared/ @balloob
 /tests/components/marantz_infrared/ @balloob
 /homeassistant/components/mastodon/ @fabaff @andrew-codechimp
diff --git a/homeassistant/components/frontend/manifest.json b/homeassistant/components/frontend/manifest.json
index 3dc2553190aa..932979620cf5 100644
--- a/homeassistant/components/frontend/manifest.json
+++ b/homeassistant/components/frontend/manifest.json
@@ -11,6 +11,7 @@
     "file_upload",
     "http",
     "lovelace",
+    "map_tiles",
     "onboarding",
     "repairs",
     "search",
diff --git a/homeassistant/components/map_tiles/__init__.py b/homeassistant/components/map_tiles/__init__.py
new file mode 100644
index 000000000000..b576174e8789
--- /dev/null
+++ b/homeassistant/components/map_tiles/__init__.py
@@ -0,0 +1,78 @@
+"""The Map tiles integration.
+
+Serves the frontend's base map - OpenStreetMap vector tiles, their TileJSON,
+glyphs and sprites, and raster tiles for devices that cannot render vector ones.
+
+A proxy is needed because the OSMF tile policy wants requests identified via
+`User-Agent` or `Referer`, and a browser can send neither: both are forbidden
+header names, and the default referrer (the page origin) would expose the
+user's Nabu Casa installation URL.
+"""
+
+from collections import deque
+from datetime import datetime
+import secrets
+from typing import Any
+
+import voluptuous as vol
+
+from homeassistant.components import websocket_api
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.helpers import config_validation as cv
+from homeassistant.helpers.event import async_track_time_interval
+from homeassistant.helpers.typing import ConfigType
+
+from .cache import MapTilesCache
+from .const import DATA_ACCESS_TOKENS, DOMAIN, TOKEN_CHANGE_INTERVAL, TOKEN_SIZE
+from .views import (
+    MapTilesGlyphsView,
+    MapTilesRasterView,
+    MapTilesSpriteIndexView,
+    MapTilesSpriteSheetView,
+    MapTilesTileJsonView,
+    MapTilesVectorView,
+)
+
+CONFIG_SCHEMA = cv.empty_config_schema(DOMAIN)
+
+
+async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
+    """Set up the Map tiles integration."""
+    # Leaflet asks for raster tiles with an <img>, which can carry no header, so
+    # the token has to live in the URL.
+    access_tokens: deque[str] = deque([secrets.token_hex(TOKEN_SIZE)], maxlen=2)
+    hass.data[DATA_ACCESS_TOKENS] = access_tokens
+
+    @callback
+    def _rotate_token(_now: datetime) -> None:
+        """Rotate the access token."""
+        access_tokens.append(secrets.token_hex(TOKEN_SIZE))
+
+    async_track_time_interval(
+        hass, _rotate_token, TOKEN_CHANGE_INTERVAL, cancel_on_shutdown=True
+    )
+
+    cache = MapTilesCache(hass)
+    for view in (
+        MapTilesTileJsonView,
+        MapTilesVectorView,
+        MapTilesRasterView,
+        MapTilesGlyphsView,
+        MapTilesSpriteIndexView,
+        MapTilesSpriteSheetView,
+    ):
+        hass.http.register_view(view(hass, cache))
+
+    websocket_api.async_register_command(hass, ws_access_token)
+    return True
+
+
+@callback
+@websocket_api.websocket_command({vol.Required("type"): "map_tiles/access_token"})
+def ws_access_token(
+    hass: HomeAssistant,
+    connection: websocket_api.ActiveConnection,
+    msg: dict[str, Any],
+) -> None:
+    """Return the current map tiles access token."""
+    connection.send_result(msg["id"], {"token": hass.data[DATA_ACCESS_TOKENS][-1]})
diff --git a/homeassistant/components/map_tiles/cache.py b/homeassistant/components/map_tiles/cache.py
new file mode 100644
index 000000000000..93d863e948d2
--- /dev/null
+++ b/homeassistant/components/map_tiles/cache.py
@@ -0,0 +1,111 @@
+"""Cache for the Map tiles integration."""
+
+import asyncio
+from collections import OrderedDict
+from collections.abc import Callable, Coroutine
+from dataclasses import dataclass
+import time
+from typing import Any, Final
+
+from homeassistant.core import HomeAssistant
+
+from .const import CACHE_MAX_BYTES, DOMAIN, MAX_CONCURRENT_FETCHES
+
+# Approximate bookkeeping cost of one entry (key, tuple, Asset, timestamp and
+# dict slot), charged so tiny bodies cannot grow the entry count without bound.
+_ENTRY_OVERHEAD: Final = 300
+
+
+@dataclass(frozen=True, slots=True)
+class Asset:
+    """An upstream response body plus the Content-Encoding it is stored in.
+
+    Kept compressed as upstream sent it, so a dense vector tile is not
+    re-compressed for every client that requests it.
+    """
+
+    body: bytes
+    encoding: str | None
+    ttl: float | None = None
+
+
+type FetchCallback = Callable[[], Coroutine[Any, Any, Asset | None]]
+
+
+def _entry_size(key: str, asset: Asset) -> int:
+    """Return what an entry counts against the size ceiling."""
+    return len(asset.body) + len(key) + _ENTRY_OVERHEAD
+
+
+class MapTilesCache:
+    """A bounded in-memory cache of upstream responses, keyed by asset path.
+
+    Entries are never dropped for being stale, so only the size ceiling evicts,
+    least recently used first.
+    """
+
+    def __init__(self, hass: HomeAssistant) -> None:
+        """Initialize the cache."""
+        self._hass = hass
+        self._max_bytes = CACHE_MAX_BYTES
+        self._entries: OrderedDict[str, tuple[Asset, float]] = OrderedDict()
+        self._size = 0
+        self._fetches: dict[str, asyncio.Task[Asset | None]] = {}
+        self._fetch_semaphore = asyncio.Semaphore(MAX_CONCURRENT_FETCHES)
+
+    async def async_get(self, key: str, ttl: int, fetch: FetchCallback) -> Asset | None:
+        """Return the entry for key, fetching or refreshing it as needed.
+
+        ttl is the fallback refresh interval, used when the stored asset carries
+        no upstream max-age of its own.
+        """
+        if (entry := self._entries.get(key)) is None:
+            return await self._async_fetch(key, fetch)
+
+        self._entries.move_to_end(key)
+        asset, stored_at = entry
+        if time.monotonic() - stored_at > (ttl if asset.ttl is None else asset.ttl):
+            # Serve the stale entry now and refresh in the background, so an
+            # upstream outage degrades to slightly old tiles, not to no map.
+            self._hass.async_create_background_task(
+                self._async_fetch(key, fetch), f"{DOMAIN} refresh {key}"
+            )
+        return asset
+
+    def _store(self, key: str, asset: Asset) -> None:
+        """Store an entry, evicting until back under the size ceiling."""
+        if (previous := self._entries.pop(key, None)) is not None:
+            self._size -= _entry_size(key, previous[0])
+
+        self._entries[key] = (asset, time.monotonic())
+        self._size += _entry_size(key, asset)
+
+        while self._size > self._max_bytes and len(self._entries) > 1:
+            evicted_key, (evicted, _stored_at) = self._entries.popitem(last=False)
+            self._size -= _entry_size(evicted_key, evicted)
+
+    async def _async_fetch(self, key: str, fetch: FetchCallback) -> Asset | None:
+        """Fetch key upstream, joining a fetch already in flight for it."""
+        if (pending := self._fetches.get(key)) is None:
+            pending = self._hass.async_create_task(
+                self._async_fetch_and_store(key, fetch), f"{DOMAIN} fetch {key}"
+            )
+            if not pending.done():
+                self._fetches[key] = pending
+                pending.add_done_callback(lambda _task: self._fetches.pop(key, None))
+
+        # Shielded: one client navigating away must not cancel the fetch the
+        # others are waiting on.
+        return await asyncio.shield(pending)
+
+    async def _async_fetch_and_store(
+        self, key: str, fetch: FetchCallback
+    ) -> Asset | None:
+        """Fetch key upstream and store what comes back."""
+        # Bounds parallel upstream requests and the in-flight body memory they
+        # hold; the store afterwards is synchronous and needs no slot.
+        async with self._fetch_semaphore:
+            asset = await fetch()
+        if asset is not None:
+            self._store(key, asset)
+        return asset
diff --git a/homeassistant/components/map_tiles/const.py b/homeassistant/components/map_tiles/const.py
new file mode 100644
index 000000000000..8add11bd3eab
--- /dev/null
+++ b/homeassistant/components/map_tiles/const.py
@@ -0,0 +1,88 @@
+"""Constants for the Map tiles integration."""
+
+from collections import deque
+from datetime import timedelta
+import re
+from typing import Final
+
+from aiohttp import ClientTimeout
+
+from homeassistant.const import __version__
+from homeassistant.util.hass_dict import HassKey
+
+DOMAIN: Final = "map_tiles"
+DATA_ACCESS_TOKENS: HassKey[deque[str]] = HassKey(DOMAIN)
+
+VECTOR_URL: Final = "https://vector.openstreetmap.org"
+RASTER_URL: Final = "https://tile.openstreetmap.org"
+TILEJSON_URL: Final = f"{VECTOR_URL}/shortbread_v1/tilejson.json"
+UPSTREAM_TIMEOUT: Final = ClientTimeout(total=10)
+
+CONTACT: Final = "abuse@home-assistant.io"
+UPSTREAM_HEADERS: Final = {
+    # OSM blocks referer-less browser requests; the accepted alternative is an
+    # identifying application `User-Agent`, which this proxy supplies because
+    # a browser cannot.
+    "User-Agent": (
+        f"HomeAssistant/{__version__} (+https://www.home-assistant.io; {CONTACT})"
+    ),
+    # Pinned to gzip so cached bodies are in an encoding every client accepts;
+    # the session default advertises whichever codecs happen to be installed.
+    "Accept-Encoding": "gzip",
+}
+
+# Fallback refresh intervals, used only when upstream sends no Cache-Control
+# max-age to honor. Intervals, not lifetimes: an expired entry is never dropped,
+# because it is what keeps the map up while upstream is unreachable.
+TILE_TTL: Final = 7 * 24 * 60 * 60
+# Glyphs and sprites are pinned to an upstream release and never change.
+ASSET_TTL: Final = 30 * 24 * 60 * 60
+# Short: the TileJSON is how upstream would announce a moved tile endpoint.
+TILEJSON_TTL: Final = 60 * 60
+
+# The OSMF asks consumers to cache tiles for at least a week; the max-age
+# delegates that to the browser cache instead of this instance's memory.
+TILE_MAX_AGE: Final = 7 * 24 * 60 * 60
+ASSET_MAX_AGE: Final = 30 * 24 * 60 * 60
+TILEJSON_MAX_AGE: Final = 5 * 60
+
+# A server side cache is needed because the access token rotates, which changes
+# every URL and empties every browser cache with it. In memory rather than on
+# disk: Home Assistant runs on SD cards, and losing the working set on restart
+# costs a few dozen requests.
+CACHE_MAX_BYTES: Final = 32 * 1024 * 1024
+
+# Far above any legitimate asset (tiles top out at a few hundred KB), so only a
+# hostile or broken upstream hits them; they bound what a single response can
+# make this process hold in memory, on the wire and after decompression.
+MAX_FETCH_BYTES: Final = 8 * 1024 * 1024
+MAX_DECOMPRESSED_BYTES: Final = 32 * 1024 * 1024
+
+# Bounds both in-flight body memory (this many concurrent fetches, each capped
+# at MAX_FETCH_BYTES) and how many parallel requests reach the volunteer-run OSM
+# servers at once.
+MAX_CONCURRENT_FETCHES: Final = 16
+
+# MapLibre overzooms above the source maxzoom, so nothing legitimate asks for a
+# vector tile past z14.
+VECTOR_MAX_ZOOM: Final = 14
+RASTER_MAX_ZOOM: Final = 19
+
+# OSM's own TileJSON omits "contributors", which their guidelines ask for.
+ATTRIBUTION: Final = (
+    '&copy; <a href="https://www.openstreetmap.org/copyright">OpenStreetMap</a>'
+    " contributors"
+)
+
+FONTSTACK_RE: Final = re.compile(
+    r"^[A-Za-z0-9 _-]{1,64}(?:,[A-Za-z0-9 _-]{1,64}){0,7}$"
+)
+GLYPH_RANGE_RE: Final = re.compile(r"^\d{1,5}-\d{1,5}\.pbf$")
+SPRITE_SET_RE: Final = re.compile(r"^[a-z0-9_-]{1,32}$")
+SPRITE_NAME_RE: Final = re.compile(r"^sprites(?:@2x)?$")
+
+# Bytes of entropy per access token.
+TOKEN_SIZE: Final = 32
+
+# Two tokens are live at a time, so one stays valid for 30 to 60 minutes.
+TOKEN_CHANGE_INTERVAL: Final = timedelta(minutes=30)
diff --git a/homeassistant/components/map_tiles/manifest.json b/homeassistant/components/map_tiles/manifest.json
new file mode 100644
index 000000000000..c14e45b1c6d7
--- /dev/null
+++ b/homeassistant/components/map_tiles/manifest.json
@@ -0,0 +1,10 @@
+{
+  "domain": "map_tiles",
+  "name": "Map tiles",
+  "codeowners": ["@home-assistant/core"],
+  "config_flow": false,
+  "dependencies": ["http", "websocket_api"],
+  "documentation": "https://www.home-assistant.io/integrations/map_tiles",
+  "integration_type": "system",
+  "quality_scale": "internal"
+}
diff --git a/homeassistant/components/map_tiles/views.py b/homeassistant/components/map_tiles/views.py
new file mode 100644
index 000000000000..e5b7880d190b
--- /dev/null
+++ b/homeassistant/components/map_tiles/views.py
@@ -0,0 +1,343 @@
+"""HTTP views for the Map tiles integration."""
+
+from functools import partial
+import gzip
+from http import HTTPStatus
+import json
+import logging
+from typing import Final, override
+import zlib
+
+from aiohttp import ClientError, hdrs, web
+
+from homeassistant.components.http import KEY_AUTHENTICATED, HomeAssistantView
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+from homeassistant.helpers.json import json_bytes
+
+from .cache import Asset, MapTilesCache
+from .const import (
+    ASSET_MAX_AGE,
+    ASSET_TTL,
+    ATTRIBUTION,
+    DATA_ACCESS_TOKENS,
+    FONTSTACK_RE,
+    GLYPH_RANGE_RE,
+    MAX_DECOMPRESSED_BYTES,
+    MAX_FETCH_BYTES,
+    RASTER_MAX_ZOOM,
+    RASTER_URL,
+    SPRITE_NAME_RE,
+    SPRITE_SET_RE,
+    TILE_MAX_AGE,
+    TILE_TTL,
+    TILEJSON_MAX_AGE,
+    TILEJSON_TTL,
+    TILEJSON_URL,
+    UPSTREAM_HEADERS,
+    UPSTREAM_TIMEOUT,
+    VECTOR_MAX_ZOOM,
+    VECTOR_URL,
+)
+
+_LOGGER = logging.getLogger(__name__)
+
+# A root-relative path works behind any reverse proxy; an absolute URL built
+# from the request's Host header would not if the proxy does not forward it.
+VECTOR_TILE_PATH = "/api/map_tiles/vector/{z}/{x}/{y}.mvt"
+
+# Cap coordinate length before int(), which is expensive on huge digit strings.
+MAX_COORDINATE_DIGITS = 8
+
+GZIP: Final = "gzip"
+
+
+def _gzip_decompress(body: bytes) -> bytes:
+    """Decompress a gzip body, refusing pathological expansion."""
+    decompressor = zlib.decompressobj(wbits=16 + zlib.MAX_WBITS)
+    decompressed = decompressor.decompress(body, MAX_DECOMPRESSED_BYTES)
+    if decompressor.unconsumed_tail:
+        raise ValueError("Decompressed body too large")
+    if not decompressor.eof or decompressor.unused_data:
+        raise ValueError("Malformed gzip body")
+    return decompressed
+
+
+def _upstream_ttl(cache_control: str) -> float | None:
+    """Return upstream's max-age in seconds, or None when it sends none."""
+    for directive in cache_control.split(","):
+        name, _, value = directive.strip().partition("=")
+        if name.lower() == "max-age" and value.isdigit():
+            return float(value)
+    return None
+
+
+class _MapTilesView(HomeAssistantView):
+    """Serve one class of map asset, from the cache or from upstream."""
+
+    requires_auth = False
+
+    content_type: str
+    ttl: int
+    max_age: int
+
+    def __init__(self, hass: HomeAssistant, cache: MapTilesCache) -> None:
+        """Initialize the view."""
+        self._hass = hass
+        self._cache = cache
+
+    def _authenticate(self, request: web.Request) -> None:
+        """Authenticate via the standard middleware or a map tiles query token."""
+        access_tokens = self._hass.data[DATA_ACCESS_TOKENS]
+        if request[KEY_AUTHENTICATED] or request.query.get("token") in access_tokens:
+            return
+        if hdrs.AUTHORIZATION in request.headers:
+            # A real Bearer attempt, so let the ban middleware count it.
+            raise web.HTTPUnauthorized
+        # Most likely a query token that expired while a dashboard sat open, so
+        # 403 rather than banning the user's own IP over it.
+        raise web.HTTPForbidden
+
+    async def _async_serve(self, key: str, url: str) -> web.Response:
+        """Serve an asset from the cache, fetching it upstream on a miss.
+
+        A gzip-encoded asset is served compressed to every client; Accept-Encoding
+        is intentionally not checked, since every browser accepts gzip. There is
+        therefore no identity variant, and hence no Vary on Accept-Encoding.
+        """
+        asset = await self._cache.async_get(
+            key, self.ttl, partial(self._async_fetch, url)
+        )
+        if asset is None:
+            return web.Response(status=HTTPStatus.BAD_GATEWAY)
+
+        headers = {hdrs.CACHE_CONTROL: f"private, max-age={self.max_age}"}
+        if asset.encoding:
+            headers[hdrs.CONTENT_ENCODING] = asset.encoding
+
+        return web.Response(
+            body=asset.body, content_type=self.content_type, headers=headers
+        )
+
+    async def _async_fetch(self, url: str) -> Asset | None:
+        """Fetch url upstream, returning None on any upstream failure."""
+        session = async_get_clientsession(self._hass)
+        # Keep the body in the encoding upstream sent, so a gzipped asset is
+        # cached compressed instead of re-compressed for every client.
+        try:
+            async with session.get(
+                url,
+                headers=UPSTREAM_HEADERS,
+                timeout=UPSTREAM_TIMEOUT,
+                auto_decompress=False,
+            ) as response:
+                if response.status >= HTTPStatus.BAD_REQUEST:
+                    _LOGGER.debug("Upstream %s returned %s", url, response.status)
+                    return None
+                # Accumulated in chunks so a hostile upstream cannot make this
+                # process buffer an arbitrarily large response. An empty body
+                # is a legitimate answer: a vector tile with nothing in it
+                # comes back as a short 200, not a 204 or a 404.
+                chunks: list[bytes] = []
+                read = 0
+                async for chunk in response.content.iter_chunked(64 * 1024):
+                    read += len(chunk)
+                    if read > MAX_FETCH_BYTES:
+                        _LOGGER.warning(
+                            "Upstream %s body exceeds %s bytes, refusing it",
+                            url,
+                            MAX_FETCH_BYTES,
+                        )
+                        return None
+                    chunks.append(chunk)
+        except (ClientError, TimeoutError) as err:
+            _LOGGER.debug("Upstream %s failed: %s", url, err)
+            return None
+
+        body = b"".join(chunks)
+        ttl = _upstream_ttl(response.headers.get(hdrs.CACHE_CONTROL, ""))
+        return Asset(body, response.headers.get(hdrs.CONTENT_ENCODING), ttl)
+
+
+class _MapTilesTileView(_MapTilesView):
+    """Serve map tiles."""
+
+    ttl = TILE_TTL
+    max_age = TILE_MAX_AGE
+    max_zoom: int
+    upstream: str
+    key_template: str
+
+    async def get(
+        self, request: web.Request, z: str, x: str, y: str
+    ) -> web.StreamResponse:
+        """Handle a GET request for a tile."""
+        self._authenticate(request)
+
+        if any(len(part) > MAX_COORDINATE_DIGITS for part in (z, x, y)):
+            return web.Response(status=HTTPStatus.NOT_FOUND)
+
+        zoom, column, row = int(z), int(x), int(y)
+        if zoom > self.max_zoom or column >= 2**zoom or row >= 2**zoom:
+            return web.Response(status=HTTPStatus.NOT_FOUND)
+
+        coordinates = {"z": zoom, "x": column, "y": row}
+        return await self._async_serve(
+            self.key_template.format(**coordinates),
+            self.upstream.format(**coordinates),
+        )
+
+
+class MapTilesVectorView(_MapTilesTileView):
+    """Serve vector tiles."""
+
+    name = "api:map_tiles:vector"
+    url = "/api/map_tiles/vector/{z:[0-9]+}/{x:[0-9]+}/{y:[0-9]+}.mvt"
+    content_type = "application/vnd.mapbox-vector-tile"
+    max_zoom = VECTOR_MAX_ZOOM
+    upstream = f"{VECTOR_URL}/shortbread_v1/{{z}}/{{x}}/{{y}}.mvt"
+    key_template = "vector/{z}/{x}/{y}.mvt"
+
+
+class MapTilesRasterView(_MapTilesTileView):
+    """Serve raster tiles, for devices that cannot render vector ones."""
+
+    name = "api:map_tiles:raster"
+    url = "/api/map_tiles/raster/{z:[0-9]+}/{x:[0-9]+}/{y:[0-9]+}.png"
+    content_type = "image/png"
+    max_zoom = RASTER_MAX_ZOOM
+    upstream = f"{RASTER_URL}/{{z}}/{{x}}/{{y}}.png"
+    key_template = "raster/{z}/{x}/{y}.png"
+
+
+class MapTilesGlyphsView(_MapTilesView):
+    """Serve the SDF glyphs the map labels are drawn from."""
+
+    name = "api:map_tiles:glyphs"
+    url = "/api/map_tiles/fonts/{fontstack}/{glyph_range}"
+    content_type = "application/x-protobuf"
+    ttl = ASSET_TTL
+    max_age = ASSET_MAX_AGE
+
+    async def get(
+        self, request: web.Request, fontstack: str, glyph_range: str
+    ) -> web.StreamResponse:
+        """Handle a GET request for a glyph range."""
+        self._authenticate(request)
+
+        if not FONTSTACK_RE.match(fontstack) or not GLYPH_RANGE_RE.match(glyph_range):
+            return web.Response(status=HTTPStatus.NOT_FOUND)
+
+        return await self._async_serve(
+            f"fonts/{fontstack}/{glyph_range}",
+            f"{VECTOR_URL}/styles/shortbread/fonts/{fontstack}/{glyph_range}",
+        )
+
+
+class _MapTilesSpritesView(_MapTilesView):
+    """Serve the icon sprites the map symbols come from."""
+
+    ttl = ASSET_TTL
+    max_age = ASSET_MAX_AGE
+    extension: str
+
+    async def get(
+        self, request: web.Request, sprite_set: str, name: str
+    ) -> web.StreamResponse:
+        """Handle a GET request for a sprite set."""
+        self._authenticate(request)
+
+        if not SPRITE_SET_RE.match(sprite_set) or not SPRITE_NAME_RE.match(name):
+            return web.Response(status=HTTPStatus.NOT_FOUND)
+
+        path = f"sprites/{sprite_set}/{name}{self.extension}"
+        return await self._async_serve(path, f"{VECTOR_URL}/styles/shortbread/{path}")
+
+
+class MapTilesSpriteIndexView(_MapTilesSpritesView):
+    """Serve the sprite index."""
+
+    name = "api:map_tiles:sprite_index"
+    url = "/api/map_tiles/sprites/{sprite_set}/{name}.json"
+    content_type = "application/json"
+    extension = ".json"
+
+
+class MapTilesSpriteSheetView(_MapTilesSpritesView):
+    """Serve the sprite sheet."""
+
+    name = "api:map_tiles:sprite_sheet"
+    url = "/api/map_tiles/sprites/{sprite_set}/{name}.png"
+    content_type = "image/png"
+    extension = ".png"
+
+
+class MapTilesTileJsonView(_MapTilesView):
+    """Serve the TileJSON, rewritten to point back at this instance."""
+
+    name = "api:map_tiles:tilejson"
+    url = "/api/map_tiles/tilejson.json"
+    content_type = "application/json"
+    ttl = TILEJSON_TTL
+    max_age = TILEJSON_MAX_AGE
+
+    async def get(self, request: web.Request) -> web.StreamResponse:
+        """Handle a GET request for the TileJSON."""
+        self._authenticate(request)
+        return await self._async_serve("tilejson.json", TILEJSON_URL)
+
+    @override
+    async def _async_fetch(self, url: str) -> Asset | None:
+        """Fetch the upstream TileJSON and republish it as ours.
+
+        The zoom range is taken from upstream (clamped to what we serve); the
+        attribution and the advertised tile endpoint are replaced with this
+        proxy's own. The tile endpoint is pinned, so the vector fetch URL does
+        not follow the upstream template.
+        """
+        if (asset := await super()._async_fetch(url)) is None:
+            return None
+        return await self._hass.async_add_executor_job(self._rebuild, asset)
+
+    def _rebuild(self, asset: Asset) -> Asset | None:
+        """Rewrite the upstream TileJSON to point back at this instance."""
+        try:
+            tilejson = json.loads(
+                _gzip_decompress(asset.body) if asset.encoding else asset.body
+            )
+        except ValueError, zlib.error:
+            _LOGGER.error("Upstream TileJSON is not valid JSON")
+            return None
+
+        if not isinstance(tilejson, dict) or not tilejson.get("tiles"):
+            _LOGGER.error("Upstream TileJSON does not list any tiles")
+            return None
+
+        try:
+            # Clamped to what the tile view will actually serve.
+            minzoom = max(int(tilejson.get("minzoom", 0)), 0)
+            maxzoom = min(
+                int(tilejson.get("maxzoom", VECTOR_MAX_ZOOM)), VECTOR_MAX_ZOOM
+            )
+        except TypeError, ValueError, OverflowError:
+            _LOGGER.error("Upstream TileJSON zoom range is not a finite number")
+            return None
+
+        # The only body built locally, so the only one this integration gzips.
+        # Kept on upstream's refresh cadence: it is how a moved endpoint arrives.
+        return Asset(
+            gzip.compress(
+                json_bytes(
+                    {
+                        **tilejson,
+                        "tiles": [VECTOR_TILE_PATH],
+                        "minzoom": minzoom,
+                        "maxzoom": maxzoom,
+                        "attribution": ATTRIBUTION,
+                    }
+                ),
+                mtime=0,
+            ),
+            GZIP,
+            asset.ttl,
+        )
diff --git a/script/hassfest/manifest.py b/script/hassfest/manifest.py
index fb23e8b3c957..b4942018234f 100644
--- a/script/hassfest/manifest.py
+++ b/script/hassfest/manifest.py
@@ -102,6 +102,7 @@ NO_IOT_CLASS = [
     "logbook",
     "logger",
     "lovelace",
+    "map_tiles",
     "media_source",
     "moisture",
     "motion",
diff --git a/script/hassfest/quality_scale.py b/script/hassfest/quality_scale.py
index 019e382a84a2..a299c596a973 100644
--- a/script/hassfest/quality_scale.py
+++ b/script/hassfest/quality_scale.py
@@ -2046,6 +2046,7 @@ NO_QUALITY_SCALE = [
     "logbook",
     "logger",
     "lovelace",
+    "map_tiles",
     "media_source",
     "moisture",
     "motion",
diff --git a/tests/components/map_tiles/__init__.py b/tests/components/map_tiles/__init__.py
new file mode 100644
index 000000000000..cfd48d64a703
--- /dev/null
+++ b/tests/components/map_tiles/__init__.py
@@ -0,0 +1 @@
+"""Tests for the Map tiles integration."""
diff --git a/tests/components/map_tiles/test_cache.py b/tests/components/map_tiles/test_cache.py
new file mode 100644
index 000000000000..ec58d9369eae
--- /dev/null
+++ b/tests/components/map_tiles/test_cache.py
@@ -0,0 +1,245 @@
+"""Tests for the Map tiles cache."""
+
+import asyncio
+from unittest.mock import patch
+
+from freezegun.api import FrozenDateTimeFactory
+import pytest
+
+from homeassistant.components.map_tiles.cache import (
+    _ENTRY_OVERHEAD,
+    Asset,
+    MapTilesCache,
+)
+from homeassistant.core import HomeAssistant
+
+_CACHE = "homeassistant.components.map_tiles.cache"
+
+TTL = 60
+
+BODY = b"0123456789"
+# What one entry with a single-character key counts against the ceiling.
+ENTRY_COST = len(BODY) + 1 + _ENTRY_OVERHEAD
+
+
+async def test_evicts_least_recently_used(hass: HomeAssistant) -> None:
+    """Test that the cache stays inside its ceiling, dropping the coldest first."""
+    with patch(f"{_CACHE}.CACHE_MAX_BYTES", 2 * ENTRY_COST):
+        cache = MapTilesCache(hass)
+    calls: list[str] = []
+
+    async def fetch(key: str) -> Asset:
+        calls.append(key)
+        return Asset(BODY, None)
+
+    for key in ("a", "b"):
+        await cache.async_get(key, TTL, lambda key=key: fetch(key))
+    # Reading "a" leaves "b" as the coldest entry.
+    await cache.async_get("a", TTL, lambda: fetch("a"))
+    await cache.async_get("c", TTL, lambda: fetch("c"))
+
+    assert calls == ["a", "b", "c"]
+
+    # "a" and "c" are still held; "b" was evicted to make room.
+    await cache.async_get("a", TTL, lambda: fetch("a"))
+    await cache.async_get("c", TTL, lambda: fetch("c"))
+    assert calls == ["a", "b", "c"]
+
+    await cache.async_get("b", TTL, lambda: fetch("b"))
+    assert calls == ["a", "b", "c", "b"]
+
+
+async def test_entry_larger_than_the_ceiling_is_kept(hass: HomeAssistant) -> None:
+    """Test that a tile bigger than the whole cache is still served from it."""
+    with patch(f"{_CACHE}.CACHE_MAX_BYTES", 10):
+        cache = MapTilesCache(hass)
+    calls: list[str] = []
+
+    async def fetch() -> Asset:
+        calls.append("fetched")
+        return Asset(b"a dense city tile at a low zoom", None)
+
+    await cache.async_get("big", TTL, fetch)
+    await cache.async_get("big", TTL, fetch)
+
+    assert calls == ["fetched"]
+
+
+async def test_empty_bodies_count_against_the_ceiling(hass: HomeAssistant) -> None:
+    """Test that entries with empty bodies cannot grow the cache without bound."""
+    with patch(f"{_CACHE}.CACHE_MAX_BYTES", 3 * (1 + _ENTRY_OVERHEAD)):
+        cache = MapTilesCache(hass)
+    calls: list[str] = []
+
+    async def fetch(key: str) -> Asset:
+        calls.append(key)
+        return Asset(b"", None)
+
+    for key in ("a", "b", "c", "d", "e"):
+        await cache.async_get(key, TTL, lambda key=key: fetch(key))
+
+    # The per-entry overhead pushed "a" out despite its zero-length body.
+    await cache.async_get("a", TTL, lambda: fetch("a"))
+    assert calls == ["a", "b", "c", "d", "e", "a"]
+
+
+async def test_the_encoding_is_cached_with_the_body(hass: HomeAssistant) -> None:
+    """Test that a cached asset still knows how its bytes are encoded."""
+    cache = MapTilesCache(hass)
+
+    async def fetch() -> Asset:
+        return Asset(b"compressed bytes", "gzip")
+
+    assert await cache.async_get("key", TTL, fetch) == Asset(
+        b"compressed bytes", "gzip"
+    )
+    assert await cache.async_get("key", TTL, fetch) == Asset(
+        b"compressed bytes", "gzip"
+    )
+
+
+async def test_failed_fetch_is_not_stored(hass: HomeAssistant) -> None:
+    """Test that a failure is retried rather than remembered."""
+    cache = MapTilesCache(hass)
+    calls: list[str] = []
+
+    async def fetch() -> Asset | None:
+        calls.append("fetched")
+        return None
+
+    assert await cache.async_get("key", TTL, fetch) is None
+    assert await cache.async_get("key", TTL, fetch) is None
+    assert len(calls) == 2
+
+
+async def test_concurrent_requests_share_one_fetch(hass: HomeAssistant) -> None:
+    """Test that a map view asking for one tile many times asks upstream once."""
+    cache = MapTilesCache(hass)
+    released = asyncio.Event()
+    calls: list[str] = []
+
+    async def fetch() -> Asset:
+        calls.append("fetched")
+        await released.wait()
+        return Asset(b"tile", None)
+
+    waiting = [
+        asyncio.create_task(cache.async_get("key", TTL, fetch)) for _ in range(5)
+    ]
+    await asyncio.sleep(0)
+    released.set()
+
+    assert await asyncio.gather(*waiting) == [Asset(b"tile", None)] * 5
+    assert calls == ["fetched"]
+
+
+async def test_a_cancelled_client_does_not_cancel_the_fetch(
+    hass: HomeAssistant,
+) -> None:
+    """Test that one client navigating away leaves the others their tile."""
+    cache = MapTilesCache(hass)
+    released = asyncio.Event()
+
+    async def fetch() -> Asset:
+        await released.wait()
+        return Asset(b"tile", None)
+
+    leaving = asyncio.create_task(cache.async_get("key", TTL, fetch))
+    staying = asyncio.create_task(cache.async_get("key", TTL, fetch))
+    await asyncio.sleep(0)
+
+    leaving.cancel()
+    released.set()
+
+    assert await staying == Asset(b"tile", None)
+
+
+async def test_a_cancelled_fetch_does_not_poison_the_key(hass: HomeAssistant) -> None:
+    """Test that a key can be fetched again after its fetch task was cancelled."""
+    cache = MapTilesCache(hass)
+    released = asyncio.Event()
+    calls: list[str] = []
+
+    async def fetch() -> Asset:
+        calls.append("fetched")
+        await released.wait()
+        return Asset(b"tile", None)
+
+    waiting = asyncio.create_task(cache.async_get("key", TTL, fetch))
+    await asyncio.sleep(0)
+
+    # Cancel the fetch task itself, as a shutdown would.
+    cache._fetches["key"].cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await waiting
+
+    released.set()
+    assert await cache.async_get("key", TTL, fetch) == Asset(b"tile", None)
+    assert len(calls) == 2
+
+
+async def test_stale_entry_is_refreshed_behind_the_response(
+    hass: HomeAssistant, freezer: FrozenDateTimeFactory
+) -> None:
+    """Test that an expired entry answers now and is replaced afterwards."""
+    cache = MapTilesCache(hass)
+    tiles = [b"first", b"second"]
+
+    async def fetch() -> Asset:
+        return Asset(tiles.pop(0), None)
+
+    assert await cache.async_get("key", TTL, fetch) == Asset(b"first", None)
+
+    freezer.tick(TTL + 1)
+    assert await cache.async_get("key", TTL, fetch) == Asset(b"first", None)
+    await hass.async_block_till_done()
+
+    assert await cache.async_get("key", TTL, fetch) == Asset(b"second", None)
+
+
+async def test_asset_ttl_overrides_the_fallback(
+    hass: HomeAssistant, freezer: FrozenDateTimeFactory
+) -> None:
+    """Test that an asset's own ttl decides staleness, not the caller's fallback."""
+    cache = MapTilesCache(hass)
+    tiles = [b"first", b"second"]
+
+    async def fetch() -> Asset:
+        return Asset(tiles.pop(0), None, ttl=TTL)
+
+    # The fallback is ten times the asset's own ttl, so only the latter can
+    # explain a refresh landing right after TTL elapses.
+    assert await cache.async_get("key", 10 * TTL, fetch) == Asset(b"first", None, TTL)
+
+    freezer.tick(TTL + 1)
+    assert await cache.async_get("key", 10 * TTL, fetch) == Asset(b"first", None, TTL)
+    await hass.async_block_till_done()
+
+    assert await cache.async_get("key", 10 * TTL, fetch) == Asset(b"second", None, TTL)
+
+
+async def test_concurrent_fetches_are_bounded(hass: HomeAssistant) -> None:
+    """Test that only so many upstream fetches run at once."""
+    with patch(f"{_CACHE}.MAX_CONCURRENT_FETCHES", 2):
+        cache = MapTilesCache(hass)
+    started = 0
+    release = asyncio.Event()
+
+    async def fetch() -> Asset:
+        nonlocal started
+        started += 1
+        await release.wait()
+        return Asset(b"tile", None)
+
+    tasks = [
+        asyncio.create_task(cache.async_get(key, TTL, fetch))
+        for key in ("a", "b", "c", "d", "e")
+    ]
+    # Let every task run up to the semaphore; only two get past it to fetch().
+    for _ in range(10):
+        await asyncio.sleep(0)
+    assert started == 2
+
+    release.set()
+    assert await asyncio.gather(*tasks) == [Asset(b"tile", None)] * 5
+    assert started == 5
diff --git a/tests/components/map_tiles/test_init.py b/tests/components/map_tiles/test_init.py
new file mode 100644
index 000000000000..d5413509cc8c
--- /dev/null
+++ b/tests/components/map_tiles/test_init.py
@@ -0,0 +1,650 @@
+"""Tests for the Map tiles integration."""
+
+from datetime import timedelta
+import gzip
+from http import HTTPStatus
+import math
+from unittest.mock import patch
+
+from aiohttp import ClientError
+from freezegun.api import FrozenDateTimeFactory
+import pytest
+
+from homeassistant.components.map_tiles.const import (
+    ASSET_MAX_AGE,
+    ATTRIBUTION,
+    DATA_ACCESS_TOKENS,
+    DOMAIN,
+    RASTER_URL,
+    TILE_MAX_AGE,
+    TILEJSON_URL,
+    TOKEN_CHANGE_INTERVAL,
+    VECTOR_URL,
+)
+from homeassistant.components.map_tiles.views import MapTilesVectorView
+from homeassistant.core import HomeAssistant
+from homeassistant.setup import async_setup_component
+
+from tests.common import async_fire_time_changed
+from tests.test_util.aiohttp import AiohttpClientMocker
+from tests.typing import ClientSessionGenerator, WebSocketGenerator
+
+VECTOR_TILE = b"\x1a\x0fnot-really-a-mvt"
+RASTER_TILE = b"\x89PNG\r\n\x1a\nnot-really-a-png"
+GLYPHS = b"not-really-a-glyph-range"
+SPRITES = b"\x89PNG\r\n\x1a\nnot-really-a-sprite-sheet"
+
+VECTOR_PATH = "/api/map_tiles/vector/12/2048/1361.mvt"
+RASTER_PATH = "/api/map_tiles/raster/12/2048/1361.png"
+GLYPHS_PATH = "/api/map_tiles/fonts/noto_sans_regular/0-255.pbf"
+TILEJSON_PATH = "/api/map_tiles/tilejson.json"
+
+VECTOR_UPSTREAM = f"{VECTOR_URL}/shortbread_v1/12/2048/1361.mvt"
+RASTER_UPSTREAM = f"{RASTER_URL}/12/2048/1361.png"
+GLYPHS_UPSTREAM = f"{VECTOR_URL}/styles/shortbread/fonts/noto_sans_regular/0-255.pbf"
+
+UPSTREAM_TILEJSON = {
+    "tilejson": "3.0.0",
+    "name": "shortbread",
+    "tiles": [f"{VECTOR_URL}/shortbread_v1/{{z}}/{{x}}/{{y}}.mvt"],
+    "minzoom": 0,
+    "maxzoom": 14,
+    "attribution": (
+        '&copy; <a href="https://www.openstreetmap.org/copyright">OpenStreetMap</a>'
+    ),
+    "vector_layers": [{"id": "ocean"}],
+}
+
+
+@pytest.fixture(autouse=True)
+async def setup_map_tiles(hass: HomeAssistant) -> None:
+    """Set up the integration for every test."""
+    assert await async_setup_component(hass, "http", {"http": {}})
+    assert await async_setup_component(hass, DOMAIN, {})
+
+
+async def test_vector_tile(
+    hass: HomeAssistant,
+    hass_client: ClientSessionGenerator,
+    aioclient_mock: AiohttpClientMocker,
+) -> None:
+    """Test serving a vector tile."""
+    aioclient_mock.get(VECTOR_UPSTREAM, content=VECTOR_TILE)
+
+    client = await hass_client()
+    resp = await client.get(VECTOR_PATH)
+
+    assert resp.status == HTTPStatus.OK
+    assert resp.content_type == "application/vnd.mapbox-vector-tile"
+    assert resp.headers["Cache-Control"] == f"private, max-age={TILE_MAX_AGE}"
+    assert await resp.read() == VECTOR_TILE
+
+
+async def test_raster_tile(
+    hass: HomeAssistant,
+    hass_client: ClientSessionGenerator,
+    aioclient_mock: AiohttpClientMocker,
+) -> None:
+    """Test serving a raster tile."""
+    aioclient_mock.get(RASTER_UPSTREAM, content=RASTER_TILE)
+
+    client = await hass_client()
+    resp = await client.get(RASTER_PATH)
+
+    assert resp.status == HTTPStatus.OK
+    assert resp.content_type == "image/png"
+    assert await resp.read() == RASTER_TILE
+
+
+async def test_glyphs(
+    hass: HomeAssistant,
+    hass_client: ClientSessionGenerator,
+    aioclient_mock: AiohttpClientMocker,
+) -> None:
+    """Test serving a glyph range."""
+    aioclient_mock.get(GLYPHS_UPSTREAM, content=GLYPHS)
+
+    client = await hass_client()
+    resp = await client.get(GLYPHS_PATH)
+
+    assert resp.status == HTTPStatus.OK
+    assert resp.content_type == "application/x-protobuf"
+    assert resp.headers["Cache-Control"] == f"private, max-age={ASSET_MAX_AGE}"
+    assert await resp.read() == GLYPHS
+
+
+@pytest.mark.parametrize(
+    ("name", "content_type"),
+    [
+        ("sprites.json", "application/json"),
+        ("sprites@2x.json", "application/json"),
+        ("sprites.png", "image/png"),
+        ("sprites@2x.png", "image/png"),
+    ],
+)
+async def test_sprites(
+    hass: HomeAssistant,
+    hass_client: ClientSessionGenerator,
+    aioclient_mock: AiohttpClientMocker,
+    name: str,
+    content_type: str,
+) -> None:
+    """Test serving the sprite index and sheet, at both scales."""
+    aioclient_mock.get(
+        f"{VECTOR_URL}/styles/shortbread/sprites/basics/{name}", content=SPRITES
+    )
+
+    client = await hass_client()
+    resp = await client.get(f"/api/map_tiles/sprites/basics/{name}")
+
+    assert resp.status == HTTPStatus.OK
+    assert resp.content_type == content_type
+    assert await resp.read() == SPRITES
+
+
+async def test_compressed_tile_is_handed_on_as_it_arrived(
+    hass: HomeAssistant,
+    hass_client: ClientSessionGenerator,
+    aioclient_mock: AiohttpClientMocker,
+) -> None:
+    """Test that a tile is compressed upstream once, not here for every client."""
+    tile = b"a vector tile large enough to be worth compressing" * 100
+    aioclient_mock.get(
+        VECTOR_UPSTREAM,
+        content=gzip.compress(tile),
+        headers={"Content-Encoding": "gzip"},
+    )
+
+    client = await hass_client()
+    resp = await client.get(VECTOR_PATH)
+
+    assert resp.status == HTTPStatus.OK
+    assert resp.headers["Content-Encoding"] == "gzip"
+    assert "Vary" not in resp.headers
+    # The test client decompresses on read; a stale `Content-Encoding` over
+    # already-decoded bytes would make this read fail.
+    assert await resp.read() == tile
+
+
+async def test_gzip_is_served_regardless_of_accept_encoding(
+    hass: HomeAssistant,
+    hass_client: ClientSessionGenerator,
+    aioclient_mock: AiohttpClientMocker,
+) -> None:
+    """Test that a stored gzip asset is served compressed even for an identity request."""
+    tile = b"a vector tile large enough to be worth compressing" * 100
+    aioclient_mock.get(
+        VECTOR_UPSTREAM,
+        content=gzip.compress(tile),
+        headers={"Content-Encoding": "gzip"},
+    )
+
+    client = await hass_client()
+    resp = await client.get(VECTOR_PATH, headers={"Accept-Encoding": "identity"})
+
+    assert resp.status == HTTPStatus.OK
+    assert resp.headers["Content-Encoding"] == "gzip"
+    # The test client decompresses on read regardless of what it requested.
+    assert await resp.read() == tile
+
+
+async def test_png_served_verbatim(
+    hass: HomeAssistant,
+    hass_client: ClientSessionGenerator,
+    aioclient_mock: AiohttpClientMocker,
+) -> None:
+    """Test that a raster tile is served exactly as upstream sent it."""
+    aioclient_mock.get(RASTER_UPSTREAM, content=RASTER_TILE)
+
+    client = await hass_client()
+    resp = await client.get(RASTER_PATH)
+
+    assert resp.status == HTTPStatus.OK
+    assert "Content-Encoding" not in resp.headers
+    assert await resp.read() == RASTER_TILE
+
+
+async def test_upstream_headers_identify_home_assistant(
+    hass: HomeAssistant,
+    hass_client: ClientSessionGenerator,
+    aioclient_mock: AiohttpClientMocker,
+) -> None:
+    """Test that the upstream request says who is asking, and nothing else."""
+    aioclient_mock.get(RASTER_UPSTREAM, content=RASTER_TILE)
+
+    client = await hass_client()
+    await client.get(RASTER_PATH)
+
+    headers = aioclient_mock.mock_calls[0][3]
+    assert headers["User-Agent"].startswith("HomeAssistant/")
+    assert "abuse@home-assistant.io" in headers["User-Agent"]
+    # Pinned to gzip so cached bodies are in an encoding every client accepts.
+    assert headers["Accept-Encoding"] == "gzip"
+    # A Referer would be the instance hostname, which identifies an installation.
+    assert "Referer" not in headers
+    assert "Cookie" not in headers
+    assert "X-Requested-With" not in headers
+
+
+async def test_tilejson_is_rewritten(
+    hass: HomeAssistant,
+    hass_client: ClientSessionGenerator,
+    aioclient_mock: AiohttpClientMocker,
+) -> None:
+    """Test that the TileJSON points back at this instance and credits OSM."""
+    aioclient_mock.get(TILEJSON_URL, json=UPSTREAM_TILEJSON)
+
+    client = await hass_client()
+    resp = await client.get(TILEJSON_PATH)
+
+    assert resp.status == HTTPStatus.OK
+    tilejson = await resp.json()
+    assert tilejson["tiles"] == ["/api/map_tiles/vector/{z}/{x}/{y}.mvt"]
+    assert tilejson["attribution"] == ATTRIBUTION
+    assert "contributors" in tilejson["attribution"]
+    assert tilejson["minzoom"] == 0
+    assert tilejson["maxzoom"] == 14
+    # Passed through, so a change upstream needs no release here.
+    assert tilejson["vector_layers"] == [{"id": "ocean"}]
+
+
+async def test_tilejson_maxzoom_is_clamped(
+    hass: HomeAssistant,
+    hass_client: ClientSessionGenerator,
+    aioclient_mock: AiohttpClientMocker,
+) -> None:
+    """Test that the advertised zoom range cannot exceed what is served."""
+    aioclient_mock.get(TILEJSON_URL, json={**UPSTREAM_TILEJSON, "maxzoom": 20})
+
+    client = await hass_client()
+    resp = await client.get(TILEJSON_PATH)
+
+    assert (await resp.json())["maxzoom"] == 14
+
+
+@pytest.mark.parametrize(
+    "upstream",
+    [
+        pytest.param({"text": "<html>not json</html>"}, id="not json"),
+        pytest.param({"json": {"tiles": []}}, id="no tiles"),
+        pytest.param({"json": {}}, id="empty"),
+        pytest.param({"json": ["not", "an", "object"]}, id="not an object"),
+        pytest.param(
+            {"content": b"not gzip", "headers": {"Content-Encoding": "gzip"}},
+            id="lying content encoding",
+        ),
+        pytest.param(
+            {"json": {**UPSTREAM_TILEJSON, "minzoom": "low"}},
+            id="minzoom not numeric",
+        ),
+        pytest.param(
+            {"json": {**UPSTREAM_TILEJSON, "maxzoom": None}},
+            id="maxzoom not numeric",
+        ),
+        pytest.param(
+            {"json": {**UPSTREAM_TILEJSON, "minzoom": math.inf}},
+            id="minzoom not finite",
+        ),
+    ],
+)
+async def test_tilejson_unusable(
+    hass: HomeAssistant,
+    hass_client: ClientSessionGenerator,
+    aioclient_mock: AiohttpClientMocker,
+    upstream: dict[str, object],
+) -> None:
+    """Test that a TileJSON we cannot use is refused rather than passed on."""
+    aioclient_mock.get(TILEJSON_URL, **upstream)
+
+    client = await hass_client()
+    resp = await client.get(TILEJSON_PATH)
+
+    assert resp.status == HTTPStatus.BAD_GATEWAY
+
+
+@pytest.mark.parametrize(
+    "path",
+    [
+        pytest.param("/api/map_tiles/vector/15/16384/10888.mvt", id="vector above z14"),
+        pytest.param(
+            "/api/map_tiles/raster/20/524288/348416.png", id="raster above z19"
+        ),
+        pytest.param("/api/map_tiles/vector/2/4/1.mvt", id="x outside the pyramid"),
+        pytest.param("/api/map_tiles/vector/2/1/4.mvt", id="y outside the pyramid"),
+        pytest.param("/api/map_tiles/vector/1/123456789/1.mvt", id="absurd coordinate"),
+        pytest.param("/api/map_tiles/vector/a/1/1.mvt", id="non numeric coordinate"),
+        pytest.param("/api/map_tiles/fonts/../../etc/passwd/0-255.pbf", id="traversal"),
+        pytest.param("/api/map_tiles/fonts/noto_sans/nonsense.pbf", id="bad range"),
+        pytest.param("/api/map_tiles/fonts/noto_sans/0-255.exe", id="bad extension"),
+        pytest.param("/api/map_tiles/sprites/BASICS/sprites.png", id="bad sprite set"),
+        pytest.param("/api/map_tiles/sprites/basics/other.png", id="bad sprite name"),
+    ],
+)
+async def test_rejected_before_upstream(
+    hass: HomeAssistant,
+    hass_client: ClientSessionGenerator,
+    aioclient_mock: AiohttpClientMocker,
+    path: str,
+) -> None:
+    """Test that a request we cannot serve never reaches the OSMF."""
+    client = await hass_client()
+    resp = await client.get(path)
+
+    assert resp.status == HTTPStatus.NOT_FOUND
+    assert aioclient_mock.call_count == 0
+
+
+@pytest.mark.parametrize(
+    ("path", "upstream_url", "failure"),
+    [
+        pytest.param(
+            VECTOR_PATH,
+            VECTOR_UPSTREAM,
+            {"status": HTTPStatus.INTERNAL_SERVER_ERROR},
+            id="vector 500",
+        ),
+        pytest.param(
+            VECTOR_PATH,
+            VECTOR_UPSTREAM,
+            {"status": HTTPStatus.NOT_FOUND},
+            id="vector 404",
+        ),
+        pytest.param(
+            VECTOR_PATH, VECTOR_UPSTREAM, {"exc": ClientError}, id="vector unreachable"
+        ),
+        pytest.param(
+            VECTOR_PATH, VECTOR_UPSTREAM, {"exc": TimeoutError}, id="vector timeout"
+        ),
+        pytest.param(
+            RASTER_PATH, RASTER_UPSTREAM, {"exc": ClientError}, id="raster unreachable"
+        ),
+        pytest.param(
+            TILEJSON_PATH, TILEJSON_URL, {"exc": ClientError}, id="tilejson unreachable"
+        ),
+        pytest.param(
+            GLYPHS_PATH,
+            GLYPHS_UPSTREAM,
+            {"status": HTTPStatus.INTERNAL_SERVER_ERROR},
+            id="glyphs 500",
+        ),
+    ],
+)
+async def test_upstream_failure(
+    hass: HomeAssistant,
+    hass_client: ClientSessionGenerator,
+    aioclient_mock: AiohttpClientMocker,
+    path: str,
+    upstream_url: str,
+    failure: dict[str, object],
+) -> None:
+    """Test that an upstream failure is reported rather than remembered."""
+    aioclient_mock.get(upstream_url, **failure)
+
+    client = await hass_client()
+    assert (await client.get(path)).status == HTTPStatus.BAD_GATEWAY
+    assert (await client.get(path)).status == HTTPStatus.BAD_GATEWAY
+    assert aioclient_mock.call_count == 2
+
+
+async def test_empty_vector_tile_is_served(
+    hass: HomeAssistant,
+    hass_client: ClientSessionGenerator,
+    aioclient_mock: AiohttpClientMocker,
+) -> None:
+    """Test that a tile with nothing in it is an answer, not a failure."""
+    aioclient_mock.get(VECTOR_UPSTREAM, content=b"")
+
+    client = await hass_client()
+    resp = await client.get(VECTOR_PATH)
+
+    assert resp.status == HTTPStatus.OK
+    assert await resp.read() == b""
+
+
+async def test_oversized_upstream_body_is_refused(
+    hass: HomeAssistant,
+    hass_client: ClientSessionGenerator,
+    aioclient_mock: AiohttpClientMocker,
+) -> None:
+    """Test that a body larger than the fetch cap is refused rather than held."""
+    aioclient_mock.get(VECTOR_UPSTREAM, content=b"x" * 200)
+
+    client = await hass_client()
+    with patch("homeassistant.components.map_tiles.views.MAX_FETCH_BYTES", 100):
+        assert (await client.get(VECTOR_PATH)).status == HTTPStatus.BAD_GATEWAY
+        # Not cached, so the second request has to go out again.
+        assert (await client.get(VECTOR_PATH)).status == HTTPStatus.BAD_GATEWAY
+
+    assert aioclient_mock.call_count == 2
+
+
+async def test_tilejson_expanding_past_the_cap_is_refused(
+    hass: HomeAssistant,
+    hass_client: ClientSessionGenerator,
+    aioclient_mock: AiohttpClientMocker,
+) -> None:
+    """Test that a TileJSON whose gzip body expands past the cap is refused."""
+    aioclient_mock.get(
+        TILEJSON_URL,
+        content=gzip.compress(b"0" * 1000),
+        headers={"Content-Encoding": "gzip"},
+    )
+
+    client = await hass_client()
+    with patch("homeassistant.components.map_tiles.views.MAX_DECOMPRESSED_BYTES", 100):
+        resp = await client.get(TILEJSON_PATH)
+
+    assert resp.status == HTTPStatus.BAD_GATEWAY
+
+
+async def test_cached_tile_is_not_fetched_again(
+    hass: HomeAssistant,
+    hass_client: ClientSessionGenerator,
+    aioclient_mock: AiohttpClientMocker,
+) -> None:
+    """Test that a second request for a tile is served from the cache."""
+    aioclient_mock.get(VECTOR_UPSTREAM, content=VECTOR_TILE)
+
+    client = await hass_client()
+    assert (await client.get(VECTOR_PATH)).status == HTTPStatus.OK
+    assert (await client.get(VECTOR_PATH)).status == HTTPStatus.OK
+
+    assert aioclient_mock.call_count == 1
+
+
+# Shortened so the frozen clock can pass a tile's TTL without also passing the
+# lifetime of the test client's own access token.
+SHORT_TTL = 5
+
+
+async def test_stale_tile_is_served_while_it_refreshes(
+    hass: HomeAssistant,
+    hass_client: ClientSessionGenerator,
+    aioclient_mock: AiohttpClientMocker,
+    freezer: FrozenDateTimeFactory,
+) -> None:
+    """Test that an expired tile is served now and replaced behind the response."""
+    aioclient_mock.get(VECTOR_UPSTREAM, content=VECTOR_TILE)
+
+    client = await hass_client()
+    with patch.object(MapTilesVectorView, "ttl", SHORT_TTL):
+        assert await (await client.get(VECTOR_PATH)).read() == VECTOR_TILE
+
+        freezer.tick(SHORT_TTL + 1)
+        aioclient_mock.clear_requests()
+        aioclient_mock.get(VECTOR_UPSTREAM, content=b"a newer tile")
+
+        # The stale bytes come back straight away rather than waiting on upstream.
+        assert await (await client.get(VECTOR_PATH)).read() == VECTOR_TILE
+        await hass.async_block_till_done()
+
+        assert aioclient_mock.call_count == 1
+        assert await (await client.get(VECTOR_PATH)).read() == b"a newer tile"
+
+
+async def test_stale_tile_survives_an_upstream_outage(
+    hass: HomeAssistant,
+    hass_client: ClientSessionGenerator,
+    aioclient_mock: AiohttpClientMocker,
+    freezer: FrozenDateTimeFactory,
+) -> None:
+    """Test that an unreachable upstream degrades to old tiles, not to no map."""
+    aioclient_mock.get(VECTOR_UPSTREAM, content=VECTOR_TILE)
+
+    client = await hass_client()
+    with patch.object(MapTilesVectorView, "ttl", SHORT_TTL):
+        assert await (await client.get(VECTOR_PATH)).read() == VECTOR_TILE
+
+        freezer.tick(SHORT_TTL + 1)
+        aioclient_mock.clear_requests()
+        aioclient_mock.get(VECTOR_UPSTREAM, exc=ClientError)
+
+        resp = await client.get(VECTOR_PATH)
+        await hass.async_block_till_done()
+
+        assert resp.status == HTTPStatus.OK
+        assert await resp.read() == VECTOR_TILE
+        assert await (await client.get(VECTOR_PATH)).read() == VECTOR_TILE
+
+
+async def test_upstream_max_age_drives_the_refresh(
+    hass: HomeAssistant,
+    hass_client: ClientSessionGenerator,
+    aioclient_mock: AiohttpClientMocker,
+    freezer: FrozenDateTimeFactory,
+) -> None:
+    """Test that upstream's Cache-Control max-age, not the fallback TTL, refreshes."""
+    aioclient_mock.get(
+        VECTOR_UPSTREAM, content=VECTOR_TILE, headers={"Cache-Control": "max-age=5"}
+    )
+
+    client = await hass_client()
+    assert await (await client.get(VECTOR_PATH)).read() == VECTOR_TILE
+
+    # Past upstream's 5 s max-age but far below the multi-day fallback TTL.
+    freezer.tick(6)
+    aioclient_mock.clear_requests()
+    aioclient_mock.get(VECTOR_UPSTREAM, content=b"a newer tile")
+
+    assert await (await client.get(VECTOR_PATH)).read() == VECTOR_TILE
+    await hass.async_block_till_done()
+
+    assert aioclient_mock.call_count == 1
+    assert await (await client.get(VECTOR_PATH)).read() == b"a newer tile"
+
+
+async def test_token_query_param_authenticates(
+    hass: HomeAssistant,
+    hass_client_no_auth: ClientSessionGenerator,
+    aioclient_mock: AiohttpClientMocker,
+) -> None:
+    """Test that a token in the query string authenticates, as an <img> must."""
+    aioclient_mock.get(RASTER_UPSTREAM, content=RASTER_TILE)
+
+    token = hass.data[DATA_ACCESS_TOKENS][-1]
+    client = await hass_client_no_auth()
+    resp = await client.get(f"{RASTER_PATH}?token={token}")
+
+    assert resp.status == HTTPStatus.OK
+    assert await resp.read() == RASTER_TILE
+
+
+async def test_both_live_tokens_authenticate(
+    hass: HomeAssistant,
+    hass_client_no_auth: ClientSessionGenerator,
+    aioclient_mock: AiohttpClientMocker,
+    freezer: FrozenDateTimeFactory,
+) -> None:
+    """Test that a URL minted before the last rotation still loads."""
+    aioclient_mock.get(RASTER_UPSTREAM, content=RASTER_TILE)
+    client = await hass_client_no_auth()
+
+    freezer.tick(TOKEN_CHANGE_INTERVAL + timedelta(seconds=1))
+    async_fire_time_changed(hass)
+    await hass.async_block_till_done()
+
+    tokens = hass.data[DATA_ACCESS_TOKENS]
+    assert len(tokens) == 2
+    for token in tokens:
+        resp = await client.get(f"{RASTER_PATH}?token={token}")
+        assert resp.status == HTTPStatus.OK
+
+
+async def test_rotated_out_token_is_rejected(
+    hass: HomeAssistant,
+    hass_client_no_auth: ClientSessionGenerator,
+    freezer: FrozenDateTimeFactory,
+) -> None:
+    """Test that a token stops authenticating once two rotations have passed."""
+    client = await hass_client_no_auth()
+    token = hass.data[DATA_ACCESS_TOKENS][-1]
+
+    for _ in range(2):
+        freezer.tick(TOKEN_CHANGE_INTERVAL)
+        async_fire_time_changed(hass)
+        await hass.async_block_till_done()
+
+    resp = await client.get(f"{RASTER_PATH}?token={token}")
+    assert resp.status == HTTPStatus.FORBIDDEN
+
+
+@pytest.mark.parametrize(
+    "path",
+    [VECTOR_PATH, RASTER_PATH, GLYPHS_PATH, TILEJSON_PATH],
+)
+async def test_unauthenticated_request_is_forbidden(
+    hass: HomeAssistant,
+    hass_client_no_auth: ClientSessionGenerator,
+    aioclient_mock: AiohttpClientMocker,
+    path: str,
+) -> None:
+    """Test that an internet exposed instance is not an open tile proxy."""
+    client = await hass_client_no_auth()
+    resp = await client.get(path)
+
+    assert resp.status == HTTPStatus.FORBIDDEN
+    assert aioclient_mock.call_count == 0
+
+
+async def test_invalid_token_is_forbidden(
+    hass: HomeAssistant,
+    hass_client_no_auth: ClientSessionGenerator,
+) -> None:
+    """Test that a wrong query token does not count as a failed login."""
+    client = await hass_client_no_auth()
+    resp = await client.get(f"{RASTER_PATH}?token=not-a-token")
+
+    assert resp.status == HTTPStatus.FORBIDDEN
+
+
+async def test_invalid_bearer_token_is_unauthorized(
+    hass: HomeAssistant,
+    hass_client_no_auth: ClientSessionGenerator,
+) -> None:
+    """Test that a real Bearer attempt is a 401, so the ban middleware sees it."""
+    client = await hass_client_no_auth()
+    resp = await client.get(
+        RASTER_PATH, headers={"Authorization": "Bearer not-a-token"}
+    )
+
+    assert resp.status == HTTPStatus.UNAUTHORIZED
+
+
+async def test_ws_access_token(
+    hass: HomeAssistant,
+    hass_ws_client: WebSocketGenerator,
+    freezer: FrozenDateTimeFactory,
+) -> None:
+    """Test handing the current token to the frontend."""
+    client = await hass_ws_client(hass)
+
+    await client.send_json_auto_id({"type": "map_tiles/access_token"})
+    first = (await client.receive_json())["result"]["token"]
+    assert first == hass.data[DATA_ACCESS_TOKENS][-1]
+
+    freezer.tick(TOKEN_CHANGE_INTERVAL)
+    async_fire_time_changed(hass)
+    await hass.async_block_till_done()
+
+    await client.send_json_auto_id({"type": "map_tiles/access_token"})
+    assert (await client.receive_json())["result"]["token"] != first
diff --git a/tests/snapshots/test_bootstrap.ambr b/tests/snapshots/test_bootstrap.ambr
index 9b296dffe0df..ec5caad54151 100644
--- a/tests/snapshots/test_bootstrap.ambr
+++ b/tests/snapshots/test_bootstrap.ambr
@@ -65,6 +65,7 @@
     'lock',
     'logger',
     'lovelace',
+    'map_tiles',
     'media_player',
     'media_source',
     'moisture',
@@ -175,6 +176,7 @@
     'lock',
     'logger',
     'lovelace',
+    'map_tiles',
     'media_player',
     'media_source',
     'moisture',