diff --git a/homeassistant/components/mqtt/util.py b/homeassistant/components/mqtt/util.py
index b8fca50a153..66eec1bdfe8 100644
--- a/homeassistant/components/mqtt/util.py
+++ b/homeassistant/components/mqtt/util.py
@@ -31,6 +31,15 @@ def valid_topic(value: Any) -> str:
         )
     if "\0" in value:
         raise vol.Invalid("MQTT topic name/filter must not contain null character.")
+    if any(char <= "\u001F" for char in value):
+        raise vol.Invalid("MQTT topic name/filter must not contain control characters.")
+    if any("\u007f" <= char <= "\u009F" for char in value):
+        raise vol.Invalid("MQTT topic name/filter must not contain control characters.")
+    if any("\ufdd0" <= char <= "\ufdef" for char in value):
+        raise vol.Invalid("MQTT topic name/filter must not contain non-characters.")
+    if any((ord(char) & 0xFFFF) in (0xFFFE, 0xFFFF) for char in value):
+        raise vol.Invalid("MQTT topic name/filter must not contain noncharacters.")
+
     return value
 
 
diff --git a/tests/components/mqtt/test_init.py b/tests/components/mqtt/test_init.py
index 03874ed331e..763cbfc1b71 100644
--- a/tests/components/mqtt/test_init.py
+++ b/tests/components/mqtt/test_init.py
@@ -522,11 +522,29 @@ def test_validate_topic():
 
     # Topics "SHOULD NOT" include these special characters
     # (not MUST NOT, RFC2119). The receiver MAY close the connection.
-    mqtt.util.valid_topic("\u0001")
-    mqtt.util.valid_topic("\u001F")
-    mqtt.util.valid_topic("\u009F")
-    mqtt.util.valid_topic("\u009F")
-    mqtt.util.valid_topic("\uffff")
+    # We enforce this because mosquitto does: https://github.com/eclipse/mosquitto/commit/94fdc9cb44c829ff79c74e1daa6f7d04283dfffd
+    with pytest.raises(vol.Invalid):
+        mqtt.util.valid_topic("\u0001")
+    with pytest.raises(vol.Invalid):
+        mqtt.util.valid_topic("\u001F")
+    with pytest.raises(vol.Invalid):
+        mqtt.util.valid_topic("\u007F")
+    with pytest.raises(vol.Invalid):
+        mqtt.util.valid_topic("\u009F")
+    with pytest.raises(vol.Invalid):
+        mqtt.util.valid_topic("\ufdd0")
+    with pytest.raises(vol.Invalid):
+        mqtt.util.valid_topic("\ufdef")
+    with pytest.raises(vol.Invalid):
+        mqtt.util.valid_topic("\ufffe")
+    with pytest.raises(vol.Invalid):
+        mqtt.util.valid_topic("\ufffe")
+    with pytest.raises(vol.Invalid):
+        mqtt.util.valid_topic("\uffff")
+    with pytest.raises(vol.Invalid):
+        mqtt.util.valid_topic("\U0001fffe")
+    with pytest.raises(vol.Invalid):
+        mqtt.util.valid_topic("\U0001ffff")
 
 
 def test_validate_subscribe_topic():