Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

channel: Implement new escape sequence #70

Merged
merged 4 commits into from
Jul 13, 2022
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 1 addition & 3 deletions tbot/machine/board/uboot.py
Original file line number Diff line number Diff line change
Expand Up @@ -380,9 +380,7 @@ def interactive(self) -> None:
can interactively run commands. This method is used by the
``interactive_uboot`` testcase.
"""
tbot.log.message(
f"Entering interactive shell ({tbot.log.c('CTRL+D to exit').bold}) ..."
)
tbot.log.message(f"Entering interactive shell...")

# It is important to send a space before the newline. Otherwise U-Boot
# will reexecute the last command which we definitely do not want here.
Expand Down
47 changes: 37 additions & 10 deletions tbot/machine/channel/channel.py
Original file line number Diff line number Diff line change
Expand Up @@ -1010,17 +1010,29 @@ def take(self) -> "Channel":

# interactive {{{
def attach_interactive(
self, end_magic: typing.Union[str, bytes, None] = None
self, end_magic: typing.Union[str, bytes, None] = None, ctrld_exit: bool = False
) -> None:
"""
Connect tbot's terminal to this channel.

Allows the user to interact directly with whatever this channel is
connected to.

:param str, bytes end_magic: String that, when detected, should end the
interactive session. If no ``end_magic`` is given, pressing
``CTRL-D`` will terminate the session.
The interactive session can be exited at any point by **pressing**
``CTRL+]`` **three times within 1 second**.

:param str, bytes end_magic: The ``end_magic`` parameter may be used to
define an automatic exit condition (sequence sent from the remote
side to trigger the end).
:param bool ctrld_exit: If ``True``, pressing ``CTRL-D`` will also
terminate the session immediately.

.. versionchanged:: UNRELEASED

- The escape sequence is now "Press ``CTRL-]`` three times within 1
second".
- The ``ctrld_exit`` parameter was added to restore the old
"``CTRL-D`` to exit" behavior.
"""
end_magic_bytes = (
end_magic.encode("utf-8") if isinstance(end_magic, str) else end_magic
Expand All @@ -1034,8 +1046,14 @@ def attach_interactive(
old_blacklist = self._write_blacklist
self._write_blacklist = []

escape_timestamp = None
previous: typing.Deque[int] = collections.deque(maxlen=3)

if not ctrld_exit:
tbot.log.message(
tbot.log.c("Press CTRL+] three times within 1 second to exit.").bold
)

oldtty = termios.tcgetattr(sys.stdin)
try:
tty.setraw(sys.stdin.fileno())
Expand Down Expand Up @@ -1067,13 +1085,22 @@ def attach_interactive(
if sys.stdin in r:
data = sys.stdin.buffer.read(4096)
previous.extend(data)
if end_magic is None and data == b"\x04":
break
for a, b in itertools.zip_longest(previous, b"\r~."):
if a != b:
break
else:

# Old ^D to exit behavior
if ctrld_exit and data == b"\x04":
break

# Detect whether ^] was pressed 3 times in 1 second
now = time.monotonic()
if escape_timestamp is None and 0x1D in previous:
escape_timestamp = now
elif escape_timestamp is not None:
if now < escape_timestamp + 1:
if previous.count(0x1D) >= 3:
break
else:
escape_timestamp = None

self.send(data)

sys.stdout.write("\r\n")
Expand Down
7 changes: 6 additions & 1 deletion tbot/machine/linux/ash.py
Original file line number Diff line number Diff line change
Expand Up @@ -268,6 +268,11 @@ def interactive(self) -> None:

try:
self.ch.sendline("exit")
self.ch.read_until_prompt(timeout=0.5)
try:
self.ch.read_until_prompt(timeout=0.5)
except TimeoutError:
# we might still be in the inner shell so let's try exiting again
self.ch.sendline("exit")
self.ch.read_until_prompt(timeout=0.5)
except TimeoutError:
raise Exception("Failed to reacquire shell after interactive session!")
7 changes: 6 additions & 1 deletion tbot/machine/linux/bash.py
Original file line number Diff line number Diff line change
Expand Up @@ -276,6 +276,11 @@ def interactive(self) -> None:

try:
self.ch.sendline("exit")
self.ch.read_until_prompt(timeout=0.5)
try:
self.ch.read_until_prompt(timeout=0.5)
except TimeoutError:
# we might still be in the inner shell so let's try exiting again
self.ch.sendline("exit")
self.ch.read_until_prompt(timeout=0.5)
except TimeoutError:
raise Exception("Failed to reacquire shell after interactive session!")
4 changes: 1 addition & 3 deletions tbot/machine/shell.py
Original file line number Diff line number Diff line change
Expand Up @@ -108,7 +108,5 @@ def interactive(self) -> None:
Connect tbot's stdio to this machine's channel. This will allow
interactive access to the machine.
"""
tbot.log.message(
f"Entering interactive shell ({tbot.log.c('CTRL+D to exit').bold}) ..."
)
tbot.log.message(f"Entering interactive shell...")
self.ch.attach_interactive()
2 changes: 1 addition & 1 deletion tbot_contrib/gdb.py
Original file line number Diff line number Diff line change
Expand Up @@ -103,7 +103,7 @@ def interactive(self) -> None:
tbot.log.message(
f"Entering interactive GDB shell ({tbot.log.c('CTRL+D to exit').bold}) ..."
)
self.ch.attach_interactive()
self.ch.attach_interactive(ctrld_exit=True)

self.ch.sendcontrol("C")
self.ch.read_until_prompt("(gdb) ")
Expand Down