Just make sure that external IPs does not log as internal/router IP.
In my case it’s always the router that gets banned, and i believe that is because the router routes it and the real IP is not forwarded.
I know there is a solution for it but in my case I would rather have whole HA go in to lockdown mode if someone actually tries than it banning one IP at the time.
Just remove: login_attempts_threshold: 3 from your configuration.yaml file and it stops banning entirely… (it will give a notification)
Excellent idea!
I wanted to write a request myself, and I see you brought me forward.
I really hope it is implemented.
This is awesome and exactly what I’m looking for!
I’ve installed a first Fire HD tablet in Kiosk mode in my home and it was banned out of nothing half an hour ago … so I had to disable the banlist (which is not the best idea from the point of security)
I’m trying a practical solution to promote it, I have no experience with Python.
I tried to play around a bit with the file: homeassistant/components/http/init.py
"""Support to serve the Home Assistant API as WSGI application."""
from __future__ import annotations
from ipaddress import ip_network
import logging
import os
import ssl
from typing import Any, Final, Optional, TypedDict, cast
from aiohttp import web
from aiohttp.typedefs import StrOrURL
from aiohttp.web_exceptions import HTTPMovedPermanently, HTTPRedirection
import voluptuous as vol
from homeassistant.components.network import async_get_source_ip
from homeassistant.const import EVENT_HOMEASSISTANT_STOP, SERVER_PORT
from homeassistant.core import Event, HomeAssistant
from homeassistant.helpers import storage
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.typing import ConfigType
from homeassistant.loader import bind_hass
from homeassistant.setup import async_start_setup, async_when_setup_or_start
from homeassistant.util import ssl as ssl_util
from .auth import setup_auth
from .ban import setup_bans
from .const import KEY_AUTHENTICATED, KEY_HASS, KEY_HASS_USER # noqa: F401
from .cors import setup_cors
from .forwarded import async_setup_forwarded
from .request_context import current_request, setup_request_context
from .security_filter import setup_security_filter
from .static import CACHE_HEADERS, CachingStaticResource
from .view import HomeAssistantView
from .web_runner import HomeAssistantTCPSite
DOMAIN: Final = "http"
CONF_SERVER_HOST: Final = "server_host"
CONF_SERVER_PORT: Final = "server_port"
CONF_BASE_URL: Final = "base_url"
CONF_SSL_CERTIFICATE: Final = "ssl_certificate"
CONF_SSL_PEER_CERTIFICATE: Final = "ssl_peer_certificate"
CONF_SSL_KEY: Final = "ssl_key"
CONF_CORS_ORIGINS: Final = "cors_allowed_origins"
CONF_USE_X_FORWARDED_FOR: Final = "use_x_forwarded_for"
CONF_TRUSTED_PROXIES: Final = "trusted_proxies"
CONF_LOGIN_ATTEMPTS_THRESHOLD: Final = "login_attempts_threshold"
CONF_IP_BAN_ENABLED: Final = "ip_ban_enabled"
CONF_IP_BAN_WHITELIST: Final = "ip_ban_whitelist"
CONF_SSL_PROFILE: Final = "ssl_profile"
SSL_MODERN: Final = "modern"
SSL_INTERMEDIATE: Final = "intermediate"
_LOGGER: Final = logging.getLogger(__name__)
DEFAULT_DEVELOPMENT: Final = "0"
# Cast to be able to load custom cards.
# My to be able to check url and version info.
DEFAULT_CORS: Final[list[str]] = ["https://cast.home-assistant.io"]
NO_LOGIN_ATTEMPT_THRESHOLD: Final = -1
MAX_CLIENT_SIZE: Final = 1024 ** 2 * 16
STORAGE_KEY: Final = DOMAIN
STORAGE_VERSION: Final = 1
SAVE_DELAY: Final = 180
HTTP_SCHEMA: Final = vol.All(
cv.deprecated(CONF_BASE_URL),
vol.Schema(
{
vol.Optional(CONF_SERVER_HOST): vol.All(
cv.ensure_list, vol.Length(min=1), [cv.string]
),
vol.Optional(CONF_SERVER_PORT, default=SERVER_PORT): cv.port,
vol.Optional(CONF_BASE_URL): cv.string,
vol.Optional(CONF_SSL_CERTIFICATE): cv.isfile,
vol.Optional(CONF_SSL_PEER_CERTIFICATE): cv.isfile,
vol.Optional(CONF_SSL_KEY): cv.isfile,
vol.Optional(CONF_CORS_ORIGINS, default=DEFAULT_CORS): vol.All(
cv.ensure_list, [cv.string]
),
vol.Inclusive(CONF_USE_X_FORWARDED_FOR, "proxy"): cv.boolean,
vol.Inclusive(CONF_TRUSTED_PROXIES, "proxy"): vol.All(
cv.ensure_list, [ip_network]
),
vol.Optional(
CONF_LOGIN_ATTEMPTS_THRESHOLD, default=NO_LOGIN_ATTEMPT_THRESHOLD
): vol.Any(cv.positive_int, NO_LOGIN_ATTEMPT_THRESHOLD),
vol.Inclusive(CONF_IP_BAN_ENABLED, "list"): cv.boolean,
vol.Inclusive(CONF_IP_BAN_WHITELIST, "list"): vol.All(
cv.ensure_list, [ip_network]
),
vol.Optional(CONF_SSL_PROFILE, default=SSL_MODERN): vol.In(
[SSL_INTERMEDIATE, SSL_MODERN]
),
}
),
)
CONFIG_SCHEMA: Final = vol.Schema({DOMAIN: HTTP_SCHEMA}, extra=vol.ALLOW_EXTRA)
class ConfData(TypedDict, total=False):
"""Typed dict for config data."""
server_host: list[str]
server_port: int
base_url: str
ssl_certificate: str
ssl_peer_certificate: str
ssl_key: str
cors_allowed_origins: list[str]
use_x_forwarded_for: bool
trusted_proxies: list[str]
login_attempts_threshold: int
ip_ban_enabled: bool
ip_ban_whitelist: list[str]
ssl_profile: str
@bind_hass
async def async_get_last_config(hass: HomeAssistant) -> dict | None:
"""Return the last known working config."""
store = storage.Store(hass, STORAGE_VERSION, STORAGE_KEY)
return cast(Optional[dict], await store.async_load())
class ApiConfig:
"""Configuration settings for API server."""
def __init__(
self,
local_ip: str,
host: str,
port: int,
use_ssl: bool,
) -> None:
"""Initialize a new API config object."""
self.local_ip = local_ip
self.host = host
self.port = port
self.use_ssl = use_ssl
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the HTTP API and debug interface."""
conf: ConfData | None = config.get(DOMAIN)
if conf is None:
conf = cast(ConfData, HTTP_SCHEMA({}))
server_host = conf.get(CONF_SERVER_HOST)
server_port = conf[CONF_SERVER_PORT]
ssl_certificate = conf.get(CONF_SSL_CERTIFICATE)
ssl_peer_certificate = conf.get(CONF_SSL_PEER_CERTIFICATE)
ssl_key = conf.get(CONF_SSL_KEY)
cors_origins = conf[CONF_CORS_ORIGINS]
use_x_forwarded_for = conf.get(CONF_USE_X_FORWARDED_FOR, False)
trusted_proxies = conf.get(CONF_TRUSTED_PROXIES) or []
is_ban_enabled = conf[CONF_IP_BAN_ENABLED]
ip_whitelist = conf.get(CONF_IP_BAN_WHITELIST) or []
login_threshold = conf[CONF_LOGIN_ATTEMPTS_THRESHOLD]
ssl_profile = conf[CONF_SSL_PROFILE]
server = HomeAssistantHTTP(
hass,
server_host=server_host,
server_port=server_port,
ssl_certificate=ssl_certificate,
ssl_peer_certificate=ssl_peer_certificate,
ssl_key=ssl_key,
cors_origins=cors_origins,
use_x_forwarded_for=use_x_forwarded_for,
trusted_proxies=trusted_proxies,
login_threshold=login_threshold,
is_ban_enabled=is_ban_enabled,
ip_whitelist=ip_whitelist,
ssl_profile=ssl_profile,
)
async def stop_server(event: Event) -> None:
"""Stop the server."""
await server.stop()
async def start_server(*_: Any) -> None:
"""Start the server."""
with async_start_setup(hass, ["http"]):
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, stop_server)
# We already checked it's not None.
assert conf is not None
await start_http_server_and_save_config(hass, dict(conf), server)
async_when_setup_or_start(hass, "frontend", start_server)
hass.http = server
local_ip = await async_get_source_ip(hass)
host = local_ip
if server_host is not None:
# Assume the first server host name provided as API host
host = server_host[0]
hass.config.api = ApiConfig(
local_ip, host, server_port, ssl_certificate is not None
)
return True
class HomeAssistantHTTP:
"""HTTP server for Home Assistant."""
def __init__(
self,
hass: HomeAssistant,
ssl_certificate: str | None,
ssl_peer_certificate: str | None,
ssl_key: str | None,
server_host: list[str] | None,
server_port: int,
cors_origins: list[str],
use_x_forwarded_for: bool,
trusted_proxies: list[str],
login_threshold: int,
is_ban_enabled: bool,
ip_whitelist: list[str],
ssl_profile: str,
) -> None:
"""Initialize the HTTP Home Assistant server."""
app = self.app = web.Application(
middlewares=[], client_max_size=MAX_CLIENT_SIZE
)
app[KEY_HASS] = hass
# Order matters, security filters middle ware needs to go first,
# forwarded middleware needs to go second.
setup_security_filter(app)
async_setup_forwarded(app, use_x_forwarded_for, trusted_proxies)
setup_request_context(app, current_request)
if is_ban_enabled:
setup_bans(hass, app, ip_whitelist, login_threshold)
setup_auth(hass, app)
setup_cors(app, cors_origins)
self.hass = hass
self.ssl_certificate = ssl_certificate
self.ssl_peer_certificate = ssl_peer_certificate
self.ssl_key = ssl_key
self.server_host = server_host
self.server_port = server_port
self.trusted_proxies = trusted_proxies
self.is_ban_enabled = is_ban_enabled
self.ip_whitelist = ip_whitelist
self.ssl_profile = ssl_profile
self._handler = None
self.runner: web.AppRunner | None = None
self.site: HomeAssistantTCPSite | None = None
def register_view(self, view: HomeAssistantView) -> None:
"""Register a view with the WSGI server.
The view argument must be a class that inherits from HomeAssistantView.
It is optional to instantiate it before registering; this method will
handle it either way.
"""
if isinstance(view, type):
# Instantiate the view, if needed
view = view()
if not hasattr(view, "url"):
class_name = view.__class__.__name__
raise AttributeError(f'{class_name} missing required attribute "url"')
if not hasattr(view, "name"):
class_name = view.__class__.__name__
raise AttributeError(f'{class_name} missing required attribute "name"')
view.register(self.app, self.app.router)
def register_redirect(
self,
url: str,
redirect_to: StrOrURL,
*,
redirect_exc: type[HTTPRedirection] = HTTPMovedPermanently,
) -> None:
"""Register a redirect with the server.
If given this must be either a string or callable. In case of a
callable it's called with the url adapter that triggered the match and
the values of the URL as keyword arguments and has to return the target
for the redirect, otherwise it has to be a string with placeholders in
rule syntax.
"""
async def redirect(request: web.Request) -> web.StreamResponse:
"""Redirect to location."""
# Should be instance of aiohttp.web_exceptions._HTTPMove.
raise redirect_exc(redirect_to) # type: ignore[arg-type,misc]
self.app["allow_configured_cors"](
self.app.router.add_route("GET", url, redirect)
)
def register_static_path(
self, url_path: str, path: str, cache_headers: bool = True
) -> None:
"""Register a folder or file to serve as a static path."""
if os.path.isdir(path):
if cache_headers:
resource: CachingStaticResource | web.StaticResource = (
CachingStaticResource(url_path, path)
)
else:
resource = web.StaticResource(url_path, path)
self.app.router.register_resource(resource)
self.app["allow_configured_cors"](resource)
return
async def serve_file(request: web.Request) -> web.FileResponse:
"""Serve file from disk."""
if cache_headers:
return web.FileResponse(path, headers=CACHE_HEADERS)
return web.FileResponse(path)
self.app["allow_configured_cors"](
self.app.router.add_route("GET", url_path, serve_file)
)
async def start(self) -> None:
"""Start the aiohttp server."""
context: ssl.SSLContext | None
if self.ssl_certificate:
try:
if self.ssl_profile == SSL_INTERMEDIATE:
context = ssl_util.server_context_intermediate()
else:
context = ssl_util.server_context_modern()
await self.hass.async_add_executor_job(
context.load_cert_chain, self.ssl_certificate, self.ssl_key
)
except OSError as error:
_LOGGER.error(
"Could not read SSL certificate from %s: %s",
self.ssl_certificate,
error,
)
return
if self.ssl_peer_certificate:
context.verify_mode = ssl.CERT_REQUIRED
await self.hass.async_add_executor_job(
context.load_verify_locations, self.ssl_peer_certificate
)
else:
context = None
# Aiohttp freezes apps after start so that no changes can be made.
# However in Home Assistant components can be discovered after boot.
# This will now raise a RunTimeError.
# To work around this we now prevent the router from getting frozen
# pylint: disable=protected-access
self.app._router.freeze = lambda: None # type: ignore[assignment]
self.runner = web.AppRunner(self.app)
await self.runner.setup()
self.site = HomeAssistantTCPSite(
self.runner, self.server_host, self.server_port, ssl_context=context
)
try:
await self.site.start()
except OSError as error:
_LOGGER.error(
"Failed to create HTTP server at port %d: %s", self.server_port, error
)
_LOGGER.info("Now listening on port %d", self.server_port)
async def stop(self) -> None:
"""Stop the aiohttp server."""
if self.site is not None:
await self.site.stop()
if self.runner is not None:
await self.runner.cleanup()
async def start_http_server_and_save_config(
hass: HomeAssistant, conf: dict, server: HomeAssistantHTTP
) -> None:
"""Startup the http server and save the config."""
await server.start()
# If we are set up successful, we store the HTTP settings for safe mode.
store = storage.Store(hass, STORAGE_VERSION, STORAGE_KEY)
if CONF_TRUSTED_PROXIES in conf:
conf[CONF_TRUSTED_PROXIES] = [
str(ip.network_address) for ip in conf[CONF_TRUSTED_PROXIES]
]
store.async_delay_save(lambda: conf, SAVE_DELAY)
Can anyone help with thet?
So that PR can be submitted to expedite it
1 Like
Nice work already @shilo
Can you share the link to your forked version of homeassistant/components/http/init.py so that we can see what changes you made.
Link in the next post
You changed line 91 from __init__.py
vol.Inclusive(CONF_IP_BAN_ENABLED, default=True): cv.boolean,
To
vol.Inclusive(CONF_IP_BAN_ENABLED, "list"): cv.boolean,
I think you need to revert that back.
Have you tested this in a dev environment?
Excellent idea !
I had connection problems because of my android companion app and now is my router IP banned. The only solution left is to set ip_ban_enabled to false, which problematic from the point of security.
How can I whitelist my router ?
If your phone made the router banned then any outside attack will also be shown as router.
You’re right. I have to find a way to solve this.
I do also randomly get login attempt warnings with my router IP. Maybe a tablet or other device on my LAN are causing it.
When I test from WAN I get the correct public client IP (not the router IP). But all devices coming from my LAN subnet gets the router IP in the ban-log.
I don’t really understand why…
Oops, I fixed this.
I did not check it, I’m not sure it’s all, it seems to me that we need to change more, I do not have time to mess with it at the moment, I would love if someone could do it
Are you still experiencing this issue? I’ve not seen it happens for a long time now.
Still happening here (not IOS related)
There are also other cases (bugs, to be sure, but a whitelist would make these significantly less annoying).
Today I was browsing the media sources from the UI, and when I tried clicking a couple of files (images), I was all of a sudden banned. No login attempts (I mean, I can’t speak for what was going on under the hood), just trying to view a file in a browser tab where I was already logged in.
You could use the following g to get some more insights on what‘s going on related to authentication
- successful ones: https://github.com/custom-components/authenticated
- failed ones: Command line Sensor - Home Assistant
I created a notification automation based on those two sensors to get notified bout logins from new IPs and new failed logins.
That of course does not help to find out WHY a device created a failed login, but it helps a bit to sort things out.
I regularly see failed attempts (all from trusted devices) and rarely my devices get locked out (maybe once every 2 to 3 months), something seems to be a bit broken with the whole authentication thing in HA.
1 Like
This feature is needed as Home Assistant’s IP ban feature is all but broken currently.
I’ve witnessed all clients (iOS, Android, Chrome on PC) having valid logins get banned due to authentication failures with a threshold set as high as 45 attempts before ban. Regardless if connecting to the internal IP or external url which first goes through caddy-docker-proxy.
It seems to be related to stale cache/authentication, as it happens very reliably after performing a Home Assistant major version update if you leave a dashboard open while Home Assistant restarts the new container.
It seems to be correlated to the cards utilized on the dashboards, with those that will automatically attempt reconnects (like the Frigate card) reliably banning clients quickly.