• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1# Copyright 2024 The Chromium Authors
2# Use of this source code is governed by a BSD-style license that can be
3# found in the LICENSE file.
4
5from __future__ import annotations
6
7import shlex
8import subprocess
9from typing import TYPE_CHECKING, Any, Dict, List, Mapping, Optional
10
11from crossbench.plt.arch import MachineArch
12from crossbench.plt.linux import RemoteLinuxPlatform
13from crossbench.plt.ssh import SshPlatformMixin
14
15if TYPE_CHECKING:
16  from crossbench.path import AnyPath, LocalPath
17  from crossbench.plt.base import CmdArg, CmdArgs, ListCmdArgs, Platform
18
19
20class LinuxSshPlatform(SshPlatformMixin, RemoteLinuxPlatform):
21
22  def __init__(self, host_platform: Platform, host: str, port: int,
23               ssh_port: int, ssh_user: str) -> None:
24    super().__init__(host_platform)
25    self._machine: Optional[MachineArch] = None
26    self._system_details: Optional[Dict[str, Any]] = None
27    self._cpu_details: Optional[Dict[str, Any]] = None
28    # TODO: move ssh-related code to SshPlatformMixin
29    self._host = host
30    self._port = port
31    self._ssh_port = ssh_port
32    self._ssh_user = ssh_user
33
34  @property
35  def name(self) -> str:
36    return "linux_ssh"
37
38  @property
39  def host(self) -> str:
40    return self._host
41
42  @property
43  def port(self) -> int:
44    return self._port
45
46  @property
47  def ssh_user(self) -> str:
48    return self._ssh_user
49
50  @property
51  def ssh_port(self) -> int:
52    return self._ssh_port
53
54  def _build_ssh_cmd(self, *args: CmdArg, shell=False) -> ListCmdArgs:
55    ssh_cmd: ListCmdArgs = [
56        "ssh", "-p", f"{self._ssh_port}", f"{self._ssh_user}@{self._host}"
57    ]
58    if shell:
59      ssh_cmd.append(*args)
60    else:
61      ssh_cmd.append(shlex.join(map(str, args)))
62    return ssh_cmd
63
64  def sh_stdout_bytes(self,
65                      *args: CmdArg,
66                      shell: bool = False,
67                      quiet: bool = False,
68                      stdin=None,
69                      env: Optional[Mapping[str, str]] = None,
70                      check: bool = True) -> bytes:
71    ssh_cmd: ListCmdArgs = self._build_ssh_cmd(*args, shell=shell)
72    return self._host_platform.sh_stdout_bytes(
73        *ssh_cmd, quiet=quiet, stdin=stdin, env=env, check=check)
74
75  def sh(self,
76         *args: CmdArg,
77         shell: bool = False,
78         capture_output: bool = False,
79         stdout=None,
80         stderr=None,
81         stdin=None,
82         env: Optional[Mapping[str, str]] = None,
83         quiet: bool = False,
84         check: bool = True) -> subprocess.CompletedProcess:
85    ssh_cmd: ListCmdArgs = self._build_ssh_cmd(*args, shell=shell)
86    return self._host_platform.sh(
87        *ssh_cmd,
88        capture_output=capture_output,
89        stdout=stdout,
90        stderr=stderr,
91        stdin=stdin,
92        env=env,
93        quiet=quiet,
94        check=check)
95
96  def popen(self,
97            *args: CmdArg,
98            bufsize=-1,
99            shell: bool = False,
100            stdout=None,
101            stderr=None,
102            stdin=None,
103            env: Optional[Mapping[str, str]] = None,
104            quiet: bool = False) -> subprocess.Popen:
105    ssh_cmd: ListCmdArgs = self._build_ssh_cmd(*args, shell=shell)
106    return self._host_platform.popen(
107        *ssh_cmd,
108        bufsize=bufsize,
109        shell=shell,
110        stdout=stdout,
111        stderr=stderr,
112        stdin=stdin,
113        env=env,
114        quiet=quiet)
115
116  def processes(self,
117                attrs: Optional[List[str]] = None) -> List[Dict[str, Any]]:
118    # TODO: Define a more generic method in PosixPlatform, possibly with
119    # an overridable function to generate ps command line.
120    lines = self.sh_stdout("ps", "-A", "-o", "pid,cmd").splitlines()
121    if len(lines) == 1:
122      return []
123
124    res: List[Dict[str, Any]] = []
125    for line in lines[1:]:
126      pid, name = line.split(maxsplit=1)
127      res.append({"pid": int(pid), "name": name})
128    return res
129
130  def push(self, from_path: LocalPath, to_path: AnyPath) -> AnyPath:
131    scp_cmd: CmdArgs = [
132        "scp", "-P", f"{self._ssh_port}", f"{from_path}",
133        f"{self._ssh_user}@{self._host}:{to_path}"
134    ]
135    self._host_platform.sh_stdout(*scp_cmd)
136    return to_path
137
138  def pull(self, from_path: AnyPath, to_path: LocalPath) -> LocalPath:
139    scp_cmd: CmdArgs = [
140        "scp", "-P", f"{self._ssh_port}",
141        f"{self._ssh_user}@{self._host}:{from_path}", f"{to_path}"
142    ]
143    self._host_platform.sh_stdout(*scp_cmd)
144    return to_path
145