feat: add IPv6 ipset support, add support for ipset_options#296
Conversation
Reviewer's GuideThis PR introduces full IPv6 support for firewall ipsets by validating entry address types, injecting family-specific options, and extending both library code and tests to handle IPv6 entries without mixing types. Class diagram for updated ipset creation and validation logicclassDiagram
class FirewallLib {
+validate_ipset_entries(ipset_entries, module)
+set_ipset(ipset, description, short, ipset_type, ipset_entries)
-_create_ipset(ipset, ipset_type, ipset_options)
}
class FirewallClientIPSetSettings {
+setType(ipset_type)
+setOptions(ipset_options)
}
FirewallLib --> FirewallClientIPSetSettings: uses
FirewallLib : validate_ipset_entries checks address type
FirewallLib : set_ipset injects family option for IPv6
FirewallLib : _create_ipset passes ipset_options
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
There was a problem hiding this comment.
Hey there - I've reviewed your changes - here's some feedback:
- Wrap the ipaddress.ip_interface call in validate_ipset_entries in a try/except so that invalid address formats produce a clear module.fail_json instead of an uncaught ValueError.
- The
ipset_optionslogic is duplicated in both the library (_create_ipset) and CLI (set_ipset) paths—consider refactoring into a shared helper to DRY up the code. - Add a unit test for a pure MAC‐only ipset to exercise the
check_macbranch in validate_ipset_entries and ensure MAC addresses are handled correctly.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- Wrap the ipaddress.ip_interface call in validate_ipset_entries in a try/except so that invalid address formats produce a clear module.fail_json instead of an uncaught ValueError.
- The `ipset_options` logic is duplicated in both the library (`_create_ipset`) and CLI (`set_ipset`) paths—consider refactoring into a shared helper to DRY up the code.
- Add a unit test for a pure MAC‐only ipset to exercise the `check_mac` branch in validate_ipset_entries and ensure MAC addresses are handled correctly.
## Individual Comments
### Comment 1
<location> `library/firewall_lib.py:352` </location>
<code_context>
+# Check that all of the ipset entries are ipv4, ipv6, or mac addresses
+# if not, fail the module
+# if they are, return "ipv4", "ipv6", or "mac"
+def validate_ipset_entries(ipset_entries, module):
+ addr_type = None
+ is_mixed_addr_types = False
</code_context>
<issue_to_address>
**issue (bug_risk):** Consider handling exceptions from ipaddress.ip_interface for invalid input.
If ipaddress.ip_interface receives invalid input, it raises a ValueError that is currently unhandled. Use a try/except block to catch this and call module.fail_json for consistent error reporting.
</issue_to_address>
### Comment 2
<location> `library/firewall_lib.py:1309-672` </location>
<code_context>
+ addr_type = validate_ipset_entries(ipset_entries, self.module)
+ if addr_type is None and ipset_entries:
+ self.module.fail_json(msg="Invalid IP address - " + str(ipset_entries))
+ ipset_options = []
+ if addr_type == "ipv6":
+ ipset_options = ["--option", "family=inet6"]
present = self.check_state(["present", "absent"], "ipset")
</code_context>
<issue_to_address>
**issue:** Inconsistent type for ipset_options between implementations.
Standardize ipset_options to use the same type in both implementations to avoid confusion and potential errors during future refactoring.
</issue_to_address>
### Comment 3
<location> `library/firewall_lib.py:668-671` </location>
<code_context>
def set_ipset(self, ipset, description, short, ipset_type, ipset_entries):
+ addr_type = validate_ipset_entries(ipset_entries, self.module)
+ if addr_type is None and ipset_entries:
+ self.module.fail_json(msg="Invalid IP address - " + str(ipset_entries))
+ ipset_options = {}
</code_context>
<issue_to_address>
**suggestion:** Redundant validation for addr_type after validate_ipset_entries.
Consider removing the addr_type None check, as validate_ipset_entries already handles invalid entries with module.fail_json.
```suggestion
addr_type = validate_ipset_entries(ipset_entries, self.module)
ipset_options = {}
```
</issue_to_address>
### Comment 4
<location> `tests/unit/test_firewall_lib.py:1366-1375` </location>
<code_context>
+ def test_create_ipset_ipv6(self, fw_ipset_settings_class, fw_class, am_class):
</code_context>
<issue_to_address>
**suggestion (testing):** Consider adding tests for invalid IPv6 addresses in ipset entries.
Adding a test for invalid IPv6 addresses will help verify that the module's validation logic properly rejects incorrect input.
Suggested implementation:
```python
def test_create_ipset_ipv6(self, fw_ipset_settings_class, fw_class, am_class):
am = am_class.return_value
am.params = {
"ipset": "test",
"ipset_type": "hash:ip",
"ipset_entries": ["2001:db8::1", "2001:db8::2", "2001:db8::3"],
"description": "test ipset ipv6",
"short": "Test",
"state": "present",
"permanent": True,
}
@patch("firewall_lib.HAS_FIREWALLD", True)
@patch("firewall_lib.FW_VERSION", "0.9.0", create=True)
@patch("firewall_lib.FirewallClient", create=True)
@patch("firewall_lib.FirewallClientIPSetSettings", create=True)
def test_create_ipset_invalid_ipv6(self, fw_ipset_settings_class, fw_class, am_class):
am = am_class.return_value
am.params = {
"ipset": "test",
"ipset_type": "hash:ip",
"ipset_entries": [
"2001:db8::zzzz", # Invalid hex
"2001:db8:1", # Missing colons
"2001:db8::g", # Invalid character
"12345::1", # Too many digits
"2001:db8::1::2", # Double ::
],
"description": "test ipset invalid ipv6",
"short": "Test",
"state": "present",
"permanent": True,
}
# Assuming the module raises a ValueError for invalid IPs
with pytest.raises(ValueError):
# Replace 'create_ipset' with the actual function/method being tested
create_ipset(am)
```
- If the function to create the ipset is not named `create_ipset`, replace it with the correct function/method call.
- If the module does not raise `ValueError` but uses another error handling mechanism, adjust the assertion accordingly (e.g., check for a specific error message or result).
- Ensure `pytest` is imported at the top of the file: `import pytest`
</issue_to_address>
### Comment 5
<location> `tests/unit/test_firewall_lib.py:1362-1376` </location>
<code_context>
+ @patch("firewall_lib.FirewallClientIPSetSettings", create=True)
+ def test_create_ipset_ipv6(self, fw_ipset_settings_class, fw_class, am_class):
+ am = am_class.return_value
+ am.params = {
+ "ipset": "test",
+ "ipset_type": "hash:ip",
+ "ipset_entries": ["2001:db8::1", "2001:db8::2", "2001:db8::3"],
+ "description": "test ipset ipv6",
+ "short": "Test",
+ "state": "present",
+ "permanent": True,
+ }
+
</code_context>
<issue_to_address>
**suggestion (testing):** Add a test for MAC address entries in ipset.
Please add a test case with only MAC addresses in ipset_entries to verify correct handling and validation.
```suggestion
@patch("firewall_lib.HAS_FIREWALLD", True)
@patch("firewall_lib.FW_VERSION", "0.9.0", create=True)
@patch("firewall_lib.FirewallClient", create=True)
@patch("firewall_lib.FirewallClientIPSetSettings", create=True)
def test_create_ipset_ipv6(self, fw_ipset_settings_class, fw_class, am_class):
am = am_class.return_value
am.params = {
"ipset": "test",
"ipset_type": "hash:ip",
"ipset_entries": ["2001:db8::1", "2001:db8::2", "2001:db8::3"],
"description": "test ipset ipv6",
"short": "Test",
"state": "present",
"permanent": True,
}
@patch("firewall_lib.HAS_FIREWALLD", True)
@patch("firewall_lib.FW_VERSION", "0.9.0", create=True)
@patch("firewall_lib.FirewallClient", create=True)
@patch("firewall_lib.FirewallClientIPSetSettings", create=True)
def test_create_ipset_mac(self, fw_ipset_settings_class, fw_class, am_class):
am = am_class.return_value
am.params = {
"ipset": "test_mac",
"ipset_type": "hash:mac",
"ipset_entries": [
"00:11:22:33:44:55",
"AA:BB:CC:DD:EE:FF",
"12:34:56:78:9A:BC"
],
"description": "test ipset mac",
"short": "TestMAC",
"state": "present",
"permanent": True,
}
# You may want to add assertions here to verify correct handling and validation
```
</issue_to_address>
### Comment 6
<location> `library/firewall_lib.py:667` </location>
<code_context>
def set_ipset(self, ipset, description, short, ipset_type, ipset_entries):
addr_type = validate_ipset_entries(ipset_entries, self.module)
if addr_type is None and ipset_entries:
self.module.fail_json(msg="Invalid IP address - " + str(ipset_entries))
ipset_options = {}
if addr_type == "ipv6":
ipset_options = {
"family": "inet6",
}
ipset_exists = ipset in self.fw.config().getIPSetNames()
fw_ipset = None
fw_ipset_settings = None
if ipset_exists:
fw_ipset = self.fw.config().getIPSetByName(ipset)
fw_ipset_settings = fw_ipset.getSettings()
if ipset_type and ipset_type != fw_ipset_settings.getType():
self.module.fail_json(
msg="Name conflict when creating ipset - "
"ipset %s of type %s already exists"
% (ipset, fw_ipset_settings.getType())
)
elif self.state == "present":
fw_ipset, fw_ipset_settings = self._create_ipset(
ipset, ipset_type, ipset_options
)
self.changed = True
ipset_exists = True
if self.state == "present":
if (
description is not None
and description != fw_ipset_settings.getDescription()
):
if not self.module.check_mode:
fw_ipset_settings.setDescription(description)
self.changed = True
if short is not None and short != fw_ipset_settings.getShort():
if not self.module.check_mode:
fw_ipset_settings.setShort(short)
self.changed = True
for entry in ipset_entries:
if not fw_ipset_settings.queryEntry(entry):
if not self.module.check_mode:
fw_ipset_settings.addEntry(entry)
self.changed = True
elif ipset_exists:
if ipset_entries:
for entry in ipset_entries:
if fw_ipset_settings.queryEntry(entry):
if not self.module.check_mode:
fw_ipset_settings.removeEntry(entry)
self.changed = True
else:
ipset_source_name = "ipset:%s" % ipset
bound_zone_permanent = self.fw.config().getZoneOfSource(
ipset_source_name
)
if bound_zone_permanent:
bound_zone_permanent = "permanent - %s" % bound_zone_permanent
bound_zone_runtime = self.fw.getZoneOfSource(ipset_source_name)
if bound_zone_runtime:
bound_zone_runtime = "runtime - %s" % bound_zone_runtime
if bound_zone_permanent or bound_zone_runtime:
bound_zones = " | ".join(
[
i
for i in [bound_zone_permanent, bound_zone_runtime]
if i != ""
]
)
if self.module.check_mode:
self.module.warn(
"Ensure %s is removed from all zones before attempting to remove it. Enabled zones: %s"
% (ipset_source_name, bound_zones)
)
else:
self.module.fail_json(
msg="Remove %s from all permanent and runtime zones before attempting to remove it"
% ipset_source_name
)
elif ipset_exists:
if not self.module.check_mode:
fw_ipset.remove()
ipset_exists = False
self.changed = True
if self.changed and not self.module.check_mode:
if ipset_exists:
fw_ipset.update(fw_ipset_settings)
self.need_reload = True
</code_context>
<issue_to_address>
**issue (code-quality):** We've found these issues:
- Use f-string instead of string concatenation ([`use-fstring-for-concatenation`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/use-fstring-for-concatenation/))
- Split conditional into multiple branches ([`split-or-ifs`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/split-or-ifs/))
- Merge duplicate blocks in conditional ([`merge-duplicate-blocks`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/merge-duplicate-blocks/))
- Remove redundant conditional [×2] ([`remove-redundant-if`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/remove-redundant-if/))
- Low code quality found in OnlineAPIBackend.set\_ipset - 11% ([`low-code-quality`](https://docs.sourcery.ai/Reference/Default-Rules/comments/low-code-quality/))
<br/><details><summary>Explanation</summary>
The quality score for this function is below the quality threshold of 25%.
This score is a combination of the method length, cognitive complexity and working memory.
How can you solve this?
It might be worth refactoring this function to make it shorter and more readable.
- Reduce the function length by extracting pieces of functionality out into
their own functions. This is the most important thing you can do - ideally a
function should be less than 10 lines.
- Reduce nesting, perhaps by introducing guard clauses to return early.
- Ensure that variables are tightly scoped, so that code using related concepts
sits together within the function rather than being scattered.</details>
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
| addr_type = validate_ipset_entries(ipset_entries, self.module) | ||
| if addr_type is None and ipset_entries: | ||
| self.module.fail_json(msg="Invalid IP address - " + str(ipset_entries)) | ||
| ipset_options = {} |
There was a problem hiding this comment.
suggestion: Redundant validation for addr_type after validate_ipset_entries.
Consider removing the addr_type None check, as validate_ipset_entries already handles invalid entries with module.fail_json.
| addr_type = validate_ipset_entries(ipset_entries, self.module) | |
| if addr_type is None and ipset_entries: | |
| self.module.fail_json(msg="Invalid IP address - " + str(ipset_entries)) | |
| ipset_options = {} | |
| addr_type = validate_ipset_entries(ipset_entries, self.module) | |
| ipset_options = {} |
| def test_create_ipset_ipv6(self, fw_ipset_settings_class, fw_class, am_class): | ||
| am = am_class.return_value | ||
| am.params = { | ||
| "ipset": "test", | ||
| "ipset_type": "hash:ip", | ||
| "ipset_entries": ["2001:db8::1", "2001:db8::2", "2001:db8::3"], | ||
| "description": "test ipset ipv6", | ||
| "short": "Test", | ||
| "state": "present", | ||
| "permanent": True, |
There was a problem hiding this comment.
suggestion (testing): Consider adding tests for invalid IPv6 addresses in ipset entries.
Adding a test for invalid IPv6 addresses will help verify that the module's validation logic properly rejects incorrect input.
Suggested implementation:
def test_create_ipset_ipv6(self, fw_ipset_settings_class, fw_class, am_class):
am = am_class.return_value
am.params = {
"ipset": "test",
"ipset_type": "hash:ip",
"ipset_entries": ["2001:db8::1", "2001:db8::2", "2001:db8::3"],
"description": "test ipset ipv6",
"short": "Test",
"state": "present",
"permanent": True,
}
@patch("firewall_lib.HAS_FIREWALLD", True)
@patch("firewall_lib.FW_VERSION", "0.9.0", create=True)
@patch("firewall_lib.FirewallClient", create=True)
@patch("firewall_lib.FirewallClientIPSetSettings", create=True)
def test_create_ipset_invalid_ipv6(self, fw_ipset_settings_class, fw_class, am_class):
am = am_class.return_value
am.params = {
"ipset": "test",
"ipset_type": "hash:ip",
"ipset_entries": [
"2001:db8::zzzz", # Invalid hex
"2001:db8:1", # Missing colons
"2001:db8::g", # Invalid character
"12345::1", # Too many digits
"2001:db8::1::2", # Double ::
],
"description": "test ipset invalid ipv6",
"short": "Test",
"state": "present",
"permanent": True,
}
# Assuming the module raises a ValueError for invalid IPs
with pytest.raises(ValueError):
# Replace 'create_ipset' with the actual function/method being tested
create_ipset(am)- If the function to create the ipset is not named
create_ipset, replace it with the correct function/method call. - If the module does not raise
ValueErrorbut uses another error handling mechanism, adjust the assertion accordingly (e.g., check for a specific error message or result). - Ensure
pytestis imported at the top of the file:import pytest
| @patch("firewall_lib.HAS_FIREWALLD", True) | ||
| @patch("firewall_lib.FW_VERSION", "0.9.0", create=True) | ||
| @patch("firewall_lib.FirewallClient", create=True) | ||
| @patch("firewall_lib.FirewallClientIPSetSettings", create=True) | ||
| def test_create_ipset_ipv6(self, fw_ipset_settings_class, fw_class, am_class): | ||
| am = am_class.return_value | ||
| am.params = { | ||
| "ipset": "test", | ||
| "ipset_type": "hash:ip", | ||
| "ipset_entries": ["2001:db8::1", "2001:db8::2", "2001:db8::3"], | ||
| "description": "test ipset ipv6", | ||
| "short": "Test", | ||
| "state": "present", | ||
| "permanent": True, | ||
| } |
There was a problem hiding this comment.
suggestion (testing): Add a test for MAC address entries in ipset.
Please add a test case with only MAC addresses in ipset_entries to verify correct handling and validation.
| @patch("firewall_lib.HAS_FIREWALLD", True) | |
| @patch("firewall_lib.FW_VERSION", "0.9.0", create=True) | |
| @patch("firewall_lib.FirewallClient", create=True) | |
| @patch("firewall_lib.FirewallClientIPSetSettings", create=True) | |
| def test_create_ipset_ipv6(self, fw_ipset_settings_class, fw_class, am_class): | |
| am = am_class.return_value | |
| am.params = { | |
| "ipset": "test", | |
| "ipset_type": "hash:ip", | |
| "ipset_entries": ["2001:db8::1", "2001:db8::2", "2001:db8::3"], | |
| "description": "test ipset ipv6", | |
| "short": "Test", | |
| "state": "present", | |
| "permanent": True, | |
| } | |
| @patch("firewall_lib.HAS_FIREWALLD", True) | |
| @patch("firewall_lib.FW_VERSION", "0.9.0", create=True) | |
| @patch("firewall_lib.FirewallClient", create=True) | |
| @patch("firewall_lib.FirewallClientIPSetSettings", create=True) | |
| def test_create_ipset_ipv6(self, fw_ipset_settings_class, fw_class, am_class): | |
| am = am_class.return_value | |
| am.params = { | |
| "ipset": "test", | |
| "ipset_type": "hash:ip", | |
| "ipset_entries": ["2001:db8::1", "2001:db8::2", "2001:db8::3"], | |
| "description": "test ipset ipv6", | |
| "short": "Test", | |
| "state": "present", | |
| "permanent": True, | |
| } | |
| @patch("firewall_lib.HAS_FIREWALLD", True) | |
| @patch("firewall_lib.FW_VERSION", "0.9.0", create=True) | |
| @patch("firewall_lib.FirewallClient", create=True) | |
| @patch("firewall_lib.FirewallClientIPSetSettings", create=True) | |
| def test_create_ipset_mac(self, fw_ipset_settings_class, fw_class, am_class): | |
| am = am_class.return_value | |
| am.params = { | |
| "ipset": "test_mac", | |
| "ipset_type": "hash:mac", | |
| "ipset_entries": [ | |
| "00:11:22:33:44:55", | |
| "AA:BB:CC:DD:EE:FF", | |
| "12:34:56:78:9A:BC" | |
| ], | |
| "description": "test ipset mac", | |
| "short": "TestMAC", | |
| "state": "present", | |
| "permanent": True, | |
| } | |
| # You may want to add assertions here to verify correct handling and validation |
| try: | ||
| from firewall.functions import check_mac | ||
|
|
||
| HAS_CHECK_MAC = True |
Check notice
Code scanning / CodeQL
Unused global variable Note
|
|
||
| HAS_CHECK_MAC = True | ||
| except ImportError: | ||
| HAS_CHECK_MAC = False |
Check notice
Code scanning / CodeQL
Unused global variable Note
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #296 +/- ##
==========================================
- Coverage 61.09% 58.26% -2.83%
==========================================
Files 2 2
Lines 910 1294 +384
==========================================
+ Hits 556 754 +198
- Misses 354 540 +186
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Sentry. 🚀 New features to boost your workflow:
|
441631f to
11a773e
Compare
|
[citest] |
|
[citest] |
|
[citest] |
|
[citest] |
a945d14 to
97bdfa3
Compare
|
[citest] |
|
[citest] |
b407932 to
47333e5
Compare
|
[citest] |
|
[citest_bad] |
Feature: Add support for IPv6 addresses to ipsets. You can now specify IPv6 addresses when using `ipset` in the `ipset_entries` list when using `hash:ip` or `hash:net`. You can also specify `ipset_options` which are extra key/value pairs of options for the ipset. Reason: Users need to be able to specify IPv6 addresses in `ipset` definitions, and need to be able to specify `ipset_options` for ipsets. Result: Users can specify IPv6 addresses and options in `ipset` definitions. NOTE: You cannot mix IPv4, IPv6, and MAC addresses in the same `ipset_entries` list. This is a limitation of the underlying firewalld implementation. Signed-off-by: Rich Megginson <rmeggins@redhat.com>
|
@erig0 please review - I didn't think it was right to add support for ipv6 ipset entries without also adding support for I'm not sure how to handle timeout - if the user specifies |
|
There is some problem with the latest centos-9 compose - the ha repo checksums do not match. |
|
Changes look fine to me, but I don't know this code. 🫣
Agree.
ACK.
Why can't this be supported? Can the role support runtime only addition of ipset entries? |
If it is using the runtime api ( @handle_exceptions
def addEntry(self, entry):
if "timeout" in self.settings[4] and self.settings[4]["timeout"] != "0":
raise FirewallError(errors.IPSET_WITH_TIMEOUT)
entry = normalize_ipset_entry(entry)
if entry not in self.settings[5]:
check_entry_overlaps_existing(entry, self.settings[5])
self.settings[5].append(entry)
else:
raise FirewallError(errors.ALREADY_ENABLED, entry)maybe if doing an operation that adds/removes entries from the ipset, remove the |
Hrm. That class is used for permanent configuration. If you want to add runtime only ipset entries you can use |
That is what I'm using, AFAICT. The issue is that firewalld does not like to add an ipset with entries and with timeout set. The error I get is from firewalld via the d-bus api: |
|
[citest] |
|
Reviewed and approved by customer |
feat: add IPv6 ipset support, add support for ipset_options
Feature: Add support for IPv6 addresses to ipsets. You can now specify IPv6 addresses
when using
ipsetin theipset_entrieslist when usinghash:iporhash:net. Youcan also specify
ipset_optionswhich are extra key/value pairs of options forthe ipset.
Reason: Users need to be able to specify IPv6 addresses in
ipsetdefinitions, and needto be able to specify
ipset_optionsfor ipsets.Result: Users can specify IPv6 addresses and options in
ipsetdefinitions.NOTE: You cannot mix IPv4, IPv6, and MAC addresses in the same
ipset_entrieslist.This is a limitation of the underlying firewalld implementation.
Signed-off-by: Rich Megginson rmeggins@redhat.com
Summary by Sourcery
Enable IPv6 support for ipsets by validating entry types, propagating the appropriate family option, and updating tests and documentation.
New Features:
Enhancements:
Documentation:
Tests: