| 82 | |
| 83 | |
| 84 | class _VNCServer: |
| 85 | def __init__(self, desktop: "Sandbox") -> None: |
| 86 | self.__novnc_handle: Optional[CommandHandle] = None |
| 87 | |
| 88 | self._vnc_port = 5900 |
| 89 | self._port = 6080 |
| 90 | self._novnc_auth_enabled = False |
| 91 | self._novnc_password = None |
| 92 | |
| 93 | self._url = f"https://{desktop.get_host(self._port)}/vnc.html" |
| 94 | |
| 95 | self.__desktop = desktop |
| 96 | |
| 97 | def _wait_for_port(self, port: int) -> bool: |
| 98 | return self.__desktop._wait_and_verify( |
| 99 | f'netstat -tuln | grep ":{port} "', lambda r: r.stdout.strip() != "" |
| 100 | ) |
| 101 | |
| 102 | def _check_vnc_running(self) -> bool: |
| 103 | try: |
| 104 | self.__desktop.commands.run("pgrep -x x11vnc") |
| 105 | return True |
| 106 | except CommandExitException: |
| 107 | return False |
| 108 | |
| 109 | @staticmethod |
| 110 | def _generate_password(length: int = 16) -> str: |
| 111 | import secrets |
| 112 | import string |
| 113 | |
| 114 | characters = string.ascii_letters + string.digits |
| 115 | return "".join(secrets.choice(characters) for _ in range(length)) |
| 116 | |
| 117 | def get_url( |
| 118 | self, |
| 119 | auto_connect: bool = True, |
| 120 | view_only: bool = False, |
| 121 | resize: str = "scale", |
| 122 | auth_key: Optional[str] = None, |
| 123 | ) -> str: |
| 124 | params = [] |
| 125 | if auto_connect: |
| 126 | params.append("autoconnect=true") |
| 127 | if view_only: |
| 128 | params.append("view_only=true") |
| 129 | if resize: |
| 130 | params.append(f"resize={resize}") |
| 131 | if auth_key: |
| 132 | params.append(f"password={auth_key}") |
| 133 | if params: |
| 134 | return f"{self._url}?{'&'.join(params)}" |
| 135 | return self._url |
| 136 | |
| 137 | def get_auth_key(self) -> str: |
| 138 | if not self._novnc_password: |
| 139 | raise RuntimeError( |
| 140 | "Unable to retrieve stream auth key, check if require_auth is enabled" |
| 141 | ) |