pybridge: Add initial authorize request to cockpit-beiboot

In order to use cockpit-beiboot as cockpit-ssh replacement from the
"normal" (not Client mode) login page, it needs to consider the given
username and password. cockpit-ssh sends an initial `authorize` message
for that and checks for "Basic" auth. Implement the same in
cockpit-beiboot.

Cover this in a new `TestLogin.testLoginSshBeiboot`. Once we generally
replace cockpit-ssh with cockpit-beiboot, this will get absorbed by
TestLogin and TestMultiMachine* and can be dropped again.

src/cockpit/beiboot.py

@@ -130,10 +130,18 @@ class AuthorizeResponder(ferny.InteractionResponder):
     commands = ('ferny.askpass', 'cockpit.report-exists')
     router: Router
 
-    def __init__(self, router: Router):
+    def __init__(self, router: Router, basic_password: Optional[str]):
         self.router = router
+        self.basic_password = basic_password
 
     async def do_askpass(self, messages: str, prompt: str, hint: str) -> Optional[str]:
+        if self.basic_password and 'password:' in prompt.lower():
+            logger.debug("AuthorizeResponder: sending Basic auth password for prompt %r", prompt)
+            # only send it once
+            reply = self.basic_password
+            self.basic_password = None
+            return reply
+
         if hint == 'none':
             # We have three problems here:
             #
@@ -182,9 +190,20 @@ def __init__(self, router: Router, destination: str, args: argparse.Namespace):
         super().__init__(router)
 
     async def do_connect_transport(self) -> None:
+        # do we have user/password (Basic auth) from the login page?
+        auth = await self.router.request_authorization("*")
+        ssh_opts = []
+        basic_password = None
+
+        if auth is not None and auth.startswith('Basic '):
+            user, basic_password = base64.b64decode(auth.split(' ', 1)[1]).decode().split(':', 1)
+            if user:  # this can be empty, i.e. auth is just ":"
+                logger.debug("got username %s and password from Basic auth", user)
+                ssh_opts = ['-l', user]
+
         beiboot_helper = BridgeBeibootHelper(self)
 
-        agent = ferny.InteractionAgent(AuthorizeResponder(self.router))
+        agent = ferny.InteractionAgent(AuthorizeResponder(self.router, basic_password))
         agent.add_handler(beiboot_helper)
 
         # We want to run a python interpreter somewhere...
@@ -214,11 +233,11 @@ async def do_connect_transport(self) -> None:
             host, _, port = self.destination.rpartition(':')
             # catch cases like `host:123` but not cases like `[2001:abcd::1]
             if port.isdigit():
-                host_args = ['-p', port, host]
+                ssh_opts += ['-p', port, host]
             else:
-                host_args = [self.destination]
+                ssh_opts += [self.destination]
 
-            cmd = ('ssh', *host_args, shlex.join(cmd))
+            cmd = ('ssh', *ssh_opts, shlex.join(cmd))
 
         # Running in flatpak?  Wrap command with flatpak-spawn --host
         if in_flatpak:

test/browser/run-test.sh

@@ -163,6 +163,7 @@ if [ "$PLAN" = "basic" ]; then
               TestLogin.testFailingWebsocketSafari
               TestLogin.testFailingWebsocketSafariNoCA
               TestLogin.testLogging
+              TestLogin.testLoginSshBeiboot
               TestLogin.testRaw
               TestLogin.testServer
               TestLogin.testUnsupportedBrowser

test/verify/check-static-login

@@ -955,6 +955,77 @@ matchrule = <SUBJECT>^DC=LAN,DC=COCKPIT,CN=alice$
         # by IPv6 address and port
         check_server("[::1]:22", expect_fp_ack=True)
 
+    # check cockpit-beiboot as replacement of cockpit-ssh on the login page
+    # once that becomes the default, TestMultiMachine* and the other TestLogin* cover this
+    @testlib.skipImage("needs pybridge", "rhel-8*", "centos-8*")
+    # enable this once our cockpit/ws container can beiboot
+    @testlib.skipOstree("client setup does not work with ws container")
+    def testLoginSshBeiboot(self):
+        m = self.machine
+        b = self.browser
+
+        # this matches our bots test VMs
+        my_ip = "172.27.0.15"
+
+        m.write("/etc/cockpit/cockpit.conf", """
+[Ssh-Login]
+Command = /usr/bin/env python3 -m cockpit.beiboot
+""", append=True)
+        m.start_cockpit()
+
+        def try_login(user, password, server=None):
+            b.open("/")
+            b.set_val('#login-user-input', user)
+            b.set_val('#login-password-input', password)
+            b.click("#show-other-login-options")
+            b.set_val("#server-field", server or my_ip)
+            b.click("#login-button")
+            # ack unknown host key; FIXME: this should be a proper authorize message, not a prompt
+            b.wait_in_text("#conversation-prompt", "authenticity of host")
+            b.set_val("#conversation-input", "yes")
+            b.click("#login-button")
+
+        def check_no_processes():
+            m.execute(f"while pgrep -af '[s]sh .* {my_ip}' >&2; do sleep 1; done")
+            m.execute("while pgrep -af '[c]ockpit.beiboot' >&2; do sleep 1; done")
+
+        def check_session():
+            b.wait_visible('#content')
+            b.enter_page('/system')
+            b.wait_visible('.system-information')
+            m.execute(f"pgrep -af '[s]sh -l admin .*{my_ip}'")
+            m.execute(f"pgrep -af '[c]ockpit.beiboot.*{my_ip}'")
+            b.logout()
+            check_no_processes()
+
+        # successful login through SSH
+        try_login("admin", "foobar")
+        check_session()
+
+        # wrong password
+        try_login("admin", "wrong")
+        # now the correct one
+        b.wait_text("#conversation-prompt", f"admin@{my_ip}'s password: ")
+        b.set_val("#conversation-input", "foobar")
+        b.click("#login-button")
+        check_session()
+
+        # wrong three times
+        self.allow_journal_messages(".*: Permission denied.*publickey.*password.*")
+        try_login("admin", "wrong")
+        for _ in range(2):
+            b.wait_text("#conversation-prompt", f"admin@{my_ip}'s password: ")
+            b.set_val("#conversation-input", "wrong")
+            b.click("#login-button")
+        # FIXME: better error message
+        b.wait_in_text("#login-fatal-message", "Internal error in login process")
+        check_no_processes()
+
+        # colliding usernames; user names in "Connect to:" are *not* supported,
+        # but pin down the behaviour
+        try_login("admin", "foobar", server=f"other@{my_ip}")
+        check_session()
+
 
 if __name__ == '__main__':
     testlib.test_main()