board.linux: Allow clobber on login prompt

Related to #1, at least handles clobber during login.
For resolving the issue, an additional fix is needed
for clobber during shell interaction.

Signed-off-by: Rahix <[email protected]>

CHANGELOG.md

@@ -3,6 +3,8 @@
 ## [Unreleased]
 ### Changed
 - Show any and all output that is received on the channel with `-vvv`
+- `BoardLinuxMachine` now allows the login and password prompts to
+  be clobbered
 
 
 ## [0.6.2] - 2018-11-22

tbot/machine/board/linux.py

@@ -71,15 +71,15 @@ def boot_to_shell(self, stream: typing.TextIO) -> str:
         chan = self._obtain_channel()
         output = ""
 
-        output += chan.read_until_prompt(self.login_prompt, stream=stream)
+        output += chan.read_until_prompt(
+            self.login_prompt, stream=stream, must_end=False
+        )
 
         chan.send(self.username + "\n")
-        stream.write(self.login_prompt + self.username + "\n")
-        output += self.login_prompt + self.username + "\n"
         if self.password is not None:
-            chan.read_until_prompt("word: ")
+            chan.read_until_prompt("word: ", stream=stream, must_end=False)
             chan.send(self.password + "\n")
-            stream.write("Password: ****\n")
+            stream.write("****\n")
             output += "Password: ****\n"
         time.sleep(self.login_wait)
         chan.send("\n")

tbot/machine/channel/channel.py

@@ -299,6 +299,7 @@ def read_until_prompt(
         regex: bool = False,
         stream: typing.Optional[typing.TextIO] = None,
         timeout: typing.Optional[float] = None,
+        must_end: bool = True,
     ) -> str:
         """
         Read until receiving ``prompt``.
@@ -311,13 +312,19 @@ def read_until_prompt(
         :param io.TextIOBase stream: Optional stream where ``read_until_prompt``
             should write everything received up until the prompt is detected.
         :param float timeout: Optional timeout.
+        :param bool must_end: Whether the prompt has to appear at the end of
+            the stream.
         :raises TimeoutError: If a timeout is set and this timeout is reached before
             the prompt is detected.
         :rtype: str
         :returns: Everything read up until the prompt.
         """
         start_time = time.monotonic()
-        expr = f"{prompt}$" if regex else "^$"
+        expr = None
+        if regex:
+            expr = f"{prompt}$" if must_end else prompt
+        elif not must_end:
+            expr = re.escape(prompt)
         buf = ""
 
         timeout_remaining = timeout
@@ -343,13 +350,17 @@ def read_until_prompt(
 
             buf += decoded
 
-            if (not regex and buf[-len(prompt) :] == prompt) or (
-                regex and re.search(expr, buf) is not None
+            if (expr is None and buf[-len(prompt) :] == prompt) or (
+                expr is not None and re.search(expr, buf) is not None
             ):
                 if stream:
-                    stream.write(decoded[: -len(prompt)])
+                    # Don't clip prompt if it doesn't need to be at the end
+                    if must_end:
+                        stream.write(decoded[: -len(prompt)])
+                    else:
+                        stream.write(decoded)
                 break
-            elif stream:
+            elif stream is not None:
                 stream.write(decoded)
 
             if timeout is not None:

tbot/tc/selftest/__init__.py

@@ -59,5 +59,6 @@ def selftest(lab: typing.Optional[linux.LabHost] = None,) -> None:
             selftest_board_linux_uboot,  # noqa: F405
             selftest_board_linux_standalone,  # noqa: F405
             selftest_board_linux_nopw,  # noqa: F405
+            selftest_board_linux_bad_console,  # noqa: F405
             lab=lh,
         )

tbot/tc/selftest/board_machine.py

@@ -303,3 +303,52 @@ class TestBoardLinuxStandalone(board.LinuxStandaloneMachine[TestBoard]):
                 except RuntimeError:
                     raised = True
                 assert raised
+
+
+@tbot.testcase
+def selftest_board_linux_bad_console(
+    lab: typing.Optional[tbot.selectable.LabHost] = None
+) -> None:
+    """Test linux booting standalone."""
+
+    class BadBoard(TestBoard):
+        def connect(self) -> channel.Channel:  # noqa: D102
+            return self.lh.new_channel(
+                linux.Raw(
+                    """\
+bash --norc --noediting; exit
+PS1="$"
+unset HISTFILE
+export UNAME="bad-board"
+bash --norc --noediting
+PS1=""
+unset HISTFILE
+set +o emacs
+set +o vi
+echo ""
+echo "[0.127] We will go into test mode now ..."
+echo "[0.128] Let's see if I can behave bad enough to break you"
+read -p 'bad-board login: [0.129] No clean login prompt for you';\
+sleep 0.02;\
+echo "[0.1337] Oh you though it was this easy?";\
+read -p "Password: [0.ORLY?] Password ain't clean either you fool
+It's even worse tbh";\
+sleep 0.02;\
+echo "[0.512] I have one last trick >:|";\
+sleep 0.2;\
+read -p ""\
+"""
+                )
+            )
+
+    class BadBoardLinux(board.LinuxStandaloneMachine[BadBoard]):
+        username = "root"
+        password = "toor"
+        login_wait = 0.02
+        shell = linux.shell.Bash
+
+    with lab or tbot.acquire_lab() as lh:
+        with BadBoard(lh) as b:
+            with BadBoardLinux(b) as lnx:
+                name = lnx.env("UNAME")
+                assert name == "bad-board", repr(name)