(self, runner: "Runner", command: str, **kwargs: Any)
| 202 | |
| 203 | # NOTE: this is for runner injection; see NOTE above _run(). |
| 204 | def _sudo(self, runner: "Runner", command: str, **kwargs: Any) -> Result: |
| 205 | prompt = self.config.sudo.prompt |
| 206 | password = kwargs.pop("password", self.config.sudo.password) |
| 207 | user = kwargs.pop("user", self.config.sudo.user) |
| 208 | env = kwargs.get("env", {}) |
| 209 | # TODO: allow subclassing for 'get the password' so users who REALLY |
| 210 | # want lazy runtime prompting can have it easily implemented. |
| 211 | # TODO: want to print a "cleaner" echo with just 'sudo <command>'; but |
| 212 | # hard to do as-is, obtaining config data from outside a Runner one |
| 213 | # holds is currently messy (could fix that), if instead we manually |
| 214 | # inspect the config ourselves that duplicates logic. NOTE: once we |
| 215 | # figure that out, there is an existing, would-fail-if-not-skipped test |
| 216 | # for this behavior in test/context.py. |
| 217 | # TODO: once that is done, though: how to handle "full debug" output |
| 218 | # exactly (display of actual, real full sudo command w/ -S and -p), in |
| 219 | # terms of API/config? Impl is easy, just go back to passing echo |
| 220 | # through to 'run'... |
| 221 | user_flags = "" |
| 222 | if user is not None: |
| 223 | user_flags = "-H -u {} ".format(user) |
| 224 | env_flags = "" |
| 225 | if env: |
| 226 | env_flags = "--preserve-env='{}' ".format(",".join(env.keys())) |
| 227 | command = self._prefix_commands(command) |
| 228 | cmd_str = "sudo -S -p '{}' {}{}{}".format( |
| 229 | prompt, env_flags, user_flags, command |
| 230 | ) |
| 231 | watcher = FailingResponder( |
| 232 | pattern=re.escape(prompt), |
| 233 | response="{}\n".format(password), |
| 234 | sentinel="Sorry, try again.\n", |
| 235 | ) |
| 236 | # Ensure we merge any user-specified watchers with our own. |
| 237 | # NOTE: If there are config-driven watchers, we pull those up to the |
| 238 | # kwarg level; that lets us merge cleanly without needing complex |
| 239 | # config-driven "override vs merge" semantics. |
| 240 | # TODO: if/when those semantics are implemented, use them instead. |
| 241 | # NOTE: config value for watchers defaults to an empty list; and we |
| 242 | # want to clone it to avoid actually mutating the config. |
| 243 | watchers = kwargs.pop("watchers", list(self.config.run.watchers)) |
| 244 | watchers.append(watcher) |
| 245 | try: |
| 246 | return runner.run(cmd_str, watchers=watchers, **kwargs) |
| 247 | except Failure as failure: |
| 248 | # Transmute failures driven by our FailingResponder, into auth |
| 249 | # failures - the command never even ran. |
| 250 | # TODO: wants to be a hook here for users that desire "override a |
| 251 | # bad config value for sudo.password" manual input |
| 252 | # NOTE: as noted in #294 comments, we MAY in future want to update |
| 253 | # this so run() is given ability to raise AuthFailure on its own. |
| 254 | # For now that has been judged unnecessary complexity. |
| 255 | if isinstance(failure.reason, ResponseNotAccepted): |
| 256 | # NOTE: not bothering with 'reason' here, it's pointless. |
| 257 | error = AuthFailure(result=failure.result, prompt=prompt) |
| 258 | raise error |
| 259 | # Reraise for any other error so it bubbles up normally. |
| 260 | else: |
| 261 | raise |
no test coverage detected