diff --git a/netmiko/huawei/__init__.py b/netmiko/huawei/__init__.py
index 40b818c63..57d30ec34 100644
--- a/netmiko/huawei/__init__.py
+++ b/netmiko/huawei/__init__.py
@@ -1,5 +1,13 @@
 from netmiko.huawei.huawei import HuaweiSSH, HuaweiVrpv8SSH
 from netmiko.huawei.huawei import HuaweiTelnet
 from netmiko.huawei.huawei_smartax import HuaweiSmartAXSSH
+from netmiko.huawei.huawei_ont import HuaweiONTTelnet, HuaweiONTSSH
 
-__all__ = ["HuaweiSmartAXSSH", "HuaweiSSH", "HuaweiVrpv8SSH", "HuaweiTelnet"]
+__all__ = [
+    "HuaweiSmartAXSSH",
+    "HuaweiSSH",
+    "HuaweiVrpv8SSH",
+    "HuaweiTelnet",
+    "HuaweiONTTelnet",
+    "HuaweiONTSSH",
+]
diff --git a/netmiko/huawei/huawei_ont.py b/netmiko/huawei/huawei_ont.py
new file mode 100644
index 000000000..95c62f88d
--- /dev/null
+++ b/netmiko/huawei/huawei_ont.py
@@ -0,0 +1,100 @@
+import re
+from typing import Optional
+
+from netmiko.no_config import NoConfig
+from netmiko.cisco_base_connection import CiscoBaseConnection
+from netmiko.exceptions import NetmikoAuthenticationException
+
+
+class HuaweiONTBase(NoConfig, CiscoBaseConnection):
+    prompt_pattern = r"WAP>"
+    su_prompt_pattern = r"SU_WAP>"
+
+    def session_preparation(self) -> None:
+        """Prepare the session after the connection has been established."""
+        self.ansi_escape_codes = True
+        # The _test_channel_read happens in special_login_handler()
+        self.set_base_prompt()
+
+    def save_config(
+        self, cmd: str = "save", confirm: bool = True, confirm_response: str = "y"
+    ) -> str:
+        """In Huawei ONTs, there is no save command. All changes are immediate."""
+        raise NotImplementedError("Save config is not supported on Huawei ONTs.")
+
+    def check_enable_mode(self, check_string: str = r"SU_WAP\>") -> bool:
+        """Check if the device is in su mode."""
+        return super().check_enable_mode(check_string=check_string)
+
+    def enable(
+        self,
+        cmd: str = "su",
+        pattern: str = "",
+        enable_pattern: Optional[str] = r"WAP\>",
+        check_state: bool = True,
+        re_flags: int = re.IGNORECASE,
+    ) -> str:
+        """Attempt to become super user."""
+        return super().enable(
+            cmd=cmd,
+            pattern=pattern,
+            enable_pattern=enable_pattern,
+            check_state=check_state,
+            re_flags=re_flags,
+        )
+
+    def exit_enable_mode(self, exit_command: str = "quit") -> str:
+        """Exit super user mode."""
+        return super().exit_enable_mode(exit_command=exit_command)
+
+    def cleanup(self, command: str = "logout") -> None:
+        """Attempt to logout."""
+        return super().cleanup(command=command)
+
+
+class HuaweiONTSSH(HuaweiONTBase):
+    """Huawei SSH driver."""
+
+    pass
+
+
+class HuaweiONTTelnet(HuaweiONTBase):
+    """Huawei Telnet driver."""
+
+    def telnet_login(
+        self,
+        pri_prompt_terminator: str = r"",
+        alt_prompt_terminator: str = r"",
+        username_pattern: str = r"(?:user:|username|login|Login|user name)",
+        pwd_pattern: str = r"Password:|password:",
+        delay_factor: float = 1.0,
+        max_loops: int = 20,
+    ) -> str:
+        """Telnet login for Huawei Devices"""
+        output = ""
+        return_msg = ""
+        try:
+            # Search for username pattern / send username
+            output = self.read_until_pattern(pattern=username_pattern, re_flags=re.I)
+            return_msg += output
+            self.write_channel(self.username + self.TELNET_RETURN)
+
+            # Search for password pattern / send password
+            output = self.read_until_pattern(pattern=pwd_pattern, re_flags=re.I)
+            return_msg += output
+            assert self.password is not None
+            self.write_channel(self.password + self.TELNET_RETURN)
+
+            output = self.read_until_pattern(pattern=self.prompt_pattern)
+            # Wait for prompt_pattern
+            if re.search(self.prompt_pattern, output):
+                return return_msg
+
+            # Should never be here
+            raise EOFError
+
+        except EOFError:
+            assert self.remote_conn is not None
+            self.remote_conn.close()
+            msg = f"Login failed: {self.host}"
+            raise NetmikoAuthenticationException(msg)
diff --git a/netmiko/ssh_dispatcher.py b/netmiko/ssh_dispatcher.py
index d05859aee..edbd5ea28 100755
--- a/netmiko/ssh_dispatcher.py
+++ b/netmiko/ssh_dispatcher.py
@@ -91,8 +91,9 @@
 from netmiko.genexis import GenexisSOLT33Telnet
 from netmiko.hillstone import HillstoneStoneosSSH
 from netmiko.hp import HPProcurveSSH, HPProcurveTelnet, HPComwareSSH, HPComwareTelnet
-from netmiko.huawei import HuaweiSSH, HuaweiVrpv8SSH, HuaweiTelnet
+from netmiko.huawei import HuaweiSSH, HuaweiVrpv8SSH, HuaweiTelnet, HuaweiONTTelnet
 from netmiko.huawei import HuaweiSmartAXSSH
+from netmiko.huawei import HuaweiONTSSH
 from netmiko.ipinfusion import IpInfusionOcNOSSSH, IpInfusionOcNOSTelnet
 from netmiko.juniper import JuniperSSH, JuniperTelnet, JuniperScreenOsSSH
 from netmiko.juniper import JuniperFileTransfer
@@ -251,6 +252,7 @@
     "huawei_olt": HuaweiSmartAXSSH,
     "huawei_vrp": HuaweiSSH,
     "huawei_vrpv8": HuaweiVrpv8SSH,
+    "huawei_ont": HuaweiONTSSH,
     "ipinfusion_ocnos": IpInfusionOcNOSSSH,
     "juniper": JuniperSSH,
     "juniper_junos": JuniperSSH,
@@ -360,6 +362,7 @@
 CLASS_MAPPER["hp_procurve_telnet"] = HPProcurveTelnet
 CLASS_MAPPER["hp_comware_telnet"] = HPComwareTelnet
 CLASS_MAPPER["huawei_telnet"] = HuaweiTelnet
+CLASS_MAPPER["huawei_ont_telnet"] = HuaweiONTTelnet
 CLASS_MAPPER["huawei_olt_telnet"] = HuaweiSmartAXSSH
 CLASS_MAPPER["ipinfusion_ocnos_telnet"] = IpInfusionOcNOSTelnet
 CLASS_MAPPER["juniper_junos_telnet"] = JuniperTelnet