Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_parse_error_unknown(self):
connection = Connection(uuid.uuid4(), "server", 445)
session = Session(connection, "user", "password")
api = SCMRApi(session)
with pytest.raises(SCMRException) as exc:
api._parse_error(999, "function_name")
assert str(exc.value) == "Exception calling function_name. Code: 999" \
", Msg: ERROR_UNKNOWN"
def test_scmr_exception(self):
with pytest.raises(SCMRException) as exc:
raise SCMRException("function_name", 1, "error_msg")
exc_msg = "Exception calling function_name. Code: 1, Msg: error_msg"
assert str(exc.value) == exc_msg
assert exc.value.function == "function_name"
assert exc.value.return_code == 1
assert exc.value.error_msg == "error_msg"
assert exc.value.message == exc_msg
def test_scmr_exception(self):
with pytest.raises(SCMRException) as exc:
raise SCMRException("function_name", 1, "error_msg")
exc_msg = "Exception calling function_name. Code: 1, Msg: error_msg"
assert str(exc.value) == exc_msg
assert exc.value.function == "function_name"
assert exc.value.return_code == 1
assert exc.value.error_msg == "error_msg"
assert exc.value.message == exc_msg
def test_parse_error(self):
connection = Connection(uuid.uuid4(), "server", 445)
session = Session(connection, "user", "password")
api = SCMRApi(session)
with pytest.raises(SCMRException) as exc:
api._parse_error(5, "function_name")
assert str(exc.value) == "Exception calling function_name. Code: 5" \
", Msg: ERROR_ACCESS_DENIED"
def _parse_error(self, return_code, function_name):
error_string = "ERROR_UNKNOWN"
for error_name, error_val in vars(ScmrReturnValues).items():
if isinstance(error_val, int) and error_val == return_code:
error_string = error_name
break
if not error_string.startswith("ERROR_SUCCESS"):
raise SCMRException(function_name, return_code, error_string)
win_client = client.Client(server=hostname, username=connection_username,
password=connection_password, port=port,
encrypt=encrypt)
try:
win_client.connect(timeout=connection_timeout)
except SMBAuthenticationError as exc:
module.fail_json(msg='Failed to authenticate over SMB: %s'
% to_text(exc))
except SMBResponseException as exc:
module.fail_json(msg='Received unexpected SMB response when opening '
'the connection: %s' % to_text(exc))
except PDUException as exc:
module.fail_json(msg='Received an exception with RPC PDU message: %s'
% to_text(exc))
except SCMRException as exc:
module.fail_json(msg='Received an exception when dealing with SCMR on '
'the Windows host: %s' % to_text(exc))
except (SMBException, PypsexecException) as exc:
module.fail_json(msg=to_text(exc))
except socket.error as exc:
module.fail_json(msg=to_text(exc))
# create PAExec service and run the process
result['changed'] = True
b_stdin = to_bytes(stdin, encoding='utf-8') if stdin else None
run_args = dict(
executable=executable, arguments=arguments, asynchronous=asynchronous,
load_profile=load_profile, interactive_session=interactive_session,
run_elevated=elevated, run_limited=limited,
username=process_username, password=process_password,
use_system_account=use_system, working_dir=working_directory,
def _open_service(self):
if self._handle:
return self._handle
# connect to the desired service in question
desired_access = DesiredAccess.SERVICE_QUERY_STATUS | \
DesiredAccess.SERVICE_START | \
DesiredAccess.SERVICE_STOP | \
DesiredAccess.DELETE
try:
log.info("Opening handle for Service %s" % self.name)
self._handle = self._scmr.open_service_w(self._scmr_handle,
self.name,
desired_access)
except SCMRException as exc:
if exc.return_code != \
ScmrReturnValues.ERROR_SERVICE_DOES_NOT_EXIST:
raise exc
else:
log.debug("Could not open handle for service %s as it did "
"not exist" % self.name)
sc_desired_access = DesiredAccess.SC_MANAGER_CONNECT | \
DesiredAccess.SC_MANAGER_CREATE_SERVICE | \
DesiredAccess.SC_MANAGER_ENUMERATE_SERVICE
scm_handle = scmr_api.open_sc_manager_w(server, None, sc_desired_access)
try:
svc_desired_access = DesiredAccess.SERVICE_QUERY_STATUS | \
DesiredAccess.SERVICE_START | \
DesiredAccess.SERVICE_STOP | \
DesiredAccess.DELETE
# delete and create a brand new service
try:
service_handle = scmr_api.open_service_w(scm_handle, svc_name,
svc_desired_access)
except SCMRException as exc:
# check the return code wasn't service does not exist
if exc.return_code != 1060:
raise exc
else:
# delete the service as it already exists
service_status = scmr_api.query_service_status(service_handle)
if service_status.current_state != CurrentState.SERVICE_STOPPED:
scmr_api.control_service(service_handle,
ControlCode.SERVICE_CONTROL_STOP)
scmr_api.delete_service(service_handle)
scmr_api.close_service_handle_w(service_handle)
# copy the executable across and overwrite the existing file
tree_admin = TreeConnect(session, r"\\%s\ADMIN$"
% session.connection.server_name)
tree_admin.connect()