Lockout agent command results if a token is received

Code Issues Proposed changes

This is a second attempt at securing the get command output endpoint
which could have data such as logs which could potentially have
sensitive details and information after the agent has completed
one or more actions.
Now, if a token is receieved, the agent locks out the command results
endpoint, and requires all future calls to include it.
This allows for the agent to be backwards compatible.
Special thanks go to cid for his first attempt at this, which I took
for the basis of some of the testing required.
Closes-Bug: #2086866
Co-Authored-By: cid@gr-oss.io
Change-Id: Ia39a3894ef5efaffd7e1d22cc6244059a32175ff

This commit is contained in:

Julia Kreger

2025年02月17日 12:09:46 -08:00

parent 8ab0bfbd9b

commit a6ca65201a

3 changed files with 124 additions and 0 deletions

Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL

Download Patch File Download Diff File Expand all files Collapse all files

17
ironic_python_agent/api/app.py

View File

@@ -113,6 +113,7 @@ class Application(object):

routing.Rule('/commands/', endpoint='run_command',

methods=['POST']),

])

self.security_get_token_support = False

def __call__(self, environ, start_response):

"""WSGI entry point."""

@@ -199,11 +200,27 @@ class Application(object):

status = self.agent.get_status()

return jsonify(status)

def require_agent_token_for_command(func):

def wrapper(self, request, *args, **kwargs):

token = request.args.get('agent_token', None)

if token:

# TODO(TheJulia): At some point down the road, remove the

# self.security_get_token_support flag and use the same

# decorator for the api_run_command endpoint.

self.security_get_token_support = True

if (self.security_get_token_support

and not self.agent.validate_agent_token(token)):

raise http_exc.Unauthorized('Token invalid.')

return func(self, request, *args, **kwargs)

return wrapper

@require_agent_token_for_command

def api_list_commands(self, request):

with metrics_utils.get_metrics_logger(__name__).timer('list_commands'):

results = self.agent.list_command_results()

return jsonify({'commands': results})

@require_agent_token_for_command

def api_get_command(self, request, cmd):

with metrics_utils.get_metrics_logger(__name__).timer('get_command'):

result = self.agent.get_command_result(cmd)

96
ironic_python_agent/tests/unit/test_api.py

View File

@@ -260,6 +260,32 @@ class TestIronicAPI(ironic_agent_base.IronicAgentTest):

],

}, response.json)

def test_list_commands_with_token(self):

agent_token = str('0123456789' * 10)

cmd_result = base.SyncCommandResult('do_things',

{'key': 'value'},

True,

{'test': 'result'})

self.mock_agent.list_command_results.return_value = [cmd_result]

self.mock_agent.validate_agent_token.return_value = True

response = self.get_json('/commands?agent_token=%s' % agent_token)

self.assertEqual(200, response.status_code)

self.assertEqual(1, self.mock_agent.validate_agent_token.call_count)

self.assertEqual(1, self.mock_agent.list_command_results.call_count)

def test_list_commands_with_token_invalid(self):

agent_token = str('0123456789' * 10)

self.mock_agent.validate_agent_token.return_value = False

response = self.get_json('/commands?agent_token=%s' % agent_token,

expect_errors=True)

self.assertEqual(401, response.status_code)

self.assertEqual(1, self.mock_agent.validate_agent_token.call_count)

self.assertEqual(0, self.mock_agent.list_command_results.call_count)

def test_get_command_result(self):

cmd_result = base.SyncCommandResult('do_things',

{'key': 'value'},

@@ -274,6 +300,76 @@ class TestIronicAPI(ironic_agent_base.IronicAgentTest):

data = response.json

self.assertEqual(serialized_cmd_result, data)

def test_get_command_with_token(self):

agent_token = str('0123456789' * 10)

cmd_result = base.SyncCommandResult('do_things',

{'key': 'value'},

True,

{'test': 'result'})

self.mock_agent.get_command_result.return_value = cmd_result

self.mock_agent.validate_agent_token.return_value = True

response = self.get_json(

'/commands/abc123?agent_token=%s' % agent_token)

self.assertEqual(200, response.status_code)

self.assertEqual(cmd_result.serialize(), response.json)

self.assertEqual(1, self.mock_agent.validate_agent_token.call_count)

self.assertEqual(1, self.mock_agent.get_command_result.call_count)

def test_get_command_with_token_invalid(self):

agent_token = str('0123456789' * 10)

self.mock_agent.validate_agent_token.return_value = False

response = self.get_json(

'/commands/abc123?agent_token=%s' % agent_token,

expect_errors=True)

self.assertEqual(401, response.status_code)

self.assertEqual(1, self.mock_agent.validate_agent_token.call_count)

self.assertEqual(0, self.mock_agent.get_command_result.call_count)

def test_get_command_locks_out_with_token(self):

"""Tests agent backwards compatibility and verifies upgrade lockout."""

cmd_result = base.SyncCommandResult('do_things',

{'key': 'value'},

True,

{'test': 'result'})

cmd_result.serialize()

self.mock_agent.get_command_result.return_value = cmd_result

agent_token = str('0123456789' * 10)

self.mock_agent.validate_agent_token.return_value = False

# Backwards compatible operation check.

response = self.get_json(

'/commands/abc123')

self.assertEqual(200, response.status_code)

self.assertFalse(self.app.security_get_token_support)

self.assertEqual(1, self.mock_agent.get_command_result.call_count)

self.mock_agent.reset_mock()

# Check with a newer ironic sending an agent_token upon the command.

# For context, in this case the token is wrong intentionally.

# It doesn't have to be right, but what we're testing is the

# submission of any value triggers the lockout

response = self.get_json(

'/commands/abc123?agent_token=%s' % agent_token,

expect_errors=True)

self.assertTrue(self.app.security_get_token_support)

self.assertEqual(401, response.status_code)

self.assertEqual(1, self.mock_agent.validate_agent_token.call_count)

self.assertEqual(0, self.mock_agent.get_command_result.call_count)

# Verifying the lockout is now being enforced and that agent token

# is now required by the agent.

response = self.get_json(

'/commands/abc123', expect_errors=True)

self.assertTrue(self.app.security_get_token_support)

self.assertEqual(401, response.status_code)

self.assertEqual(0, self.mock_agent.get_command_result.call_count)

# Verify we still called validate_agent_token

self.assertEqual(2, self.mock_agent.validate_agent_token.call_count)

def test_execute_agent_command_with_token(self):

agent_token = str('0123456789' * 10)

command = {

11
releasenotes/notes/lockout-command-result-a368187515385270.yaml Normal file

View File

@@ -0,0 +1,11 @@

---

fixes:

- |

Fixes a potential security issue where a third party may be able to

retrieve potentially sensitive data in command result output from

the agent. If a request comes in with an ``agent_token`` to the

command results endpoint, the agent will now require all future

calls to leverage the token to retrieve results and validate

that token's validity. This effectively eliminates the possibility

of a malicious entity with access to the agent's API endpoint from

capturing the command results from agent operations.