Modified firewall rule to add policy comparisons and added tests. #686
Conversation
There was a problem hiding this comment.
This is going to be great, I'd like to see similar functionality for all base resource types. Hopefully some concepts can be generalized so that it doesn't take as much code for each resource type that we want to compare at this level. Granted firewall rules are a bit more complex than most resource types.
| self.target_tags == other.target_tags and | ||
| self.source_ranges == other.source_ranges and | ||
| self.destination_ranges == other.destination_ranges and | ||
| self.allowed == other.allowed and |
There was a problem hiding this comment.
Instead of comparing self.allowed and self.denied, you should use the new self.firewall_action == other.firewall_action.
It would also make sense to have an is_equivalent method for FirewallRule as well.
| """ | ||
| return (self.direction == other.direction and | ||
| self.network == other.network and | ||
| self.source_tags == other.source_tags and |
There was a problem hiding this comment.
equivalence for lists should wrap them in a set since the lists could be unordered, or wrap them in sorted(), so frozenset(self.source_tags) == frozenset(other.source_tags). Same for the other list types. Is there any reason to not store them as a frozenset to begin with?
Something like:
def init(...):
...
self._source_tags = frozenset(parser.json_unstringify(kwargs.get('firewall_rule_source_tags', '[]')))
...
@Property
def source_tags(self):
return sorted(self._source_tags)
def eq(self):
...
self._source_tags == other._source_tags and
...
| self.rules.keys() == other.rules.keys() and | ||
| all([ | ||
| self.ports_are_equal( | ||
| self.rules.get(protocol, []), |
There was a problem hiding this comment.
This should use expanded_rules in place of rules for all comparisons.
| self._applies_to_all = True | ||
| return self._applies_to_all | ||
| return self._applies_to_all | ||
|
|
There was a problem hiding this comment.
nit: remove extra line break.
| Returns: | ||
| bool: If this action is the exact same as the other FirewallAction. | ||
| """ | ||
| return self.action == other.action and self.rules == other.rules |
There was a problem hiding this comment.
The dicts in self.rules and other.rules could be in different orders, and the ports could be in different orders.
What about in init() doing something like this:
self.rules = sort_rules(firewall_rule_allowed)
...
def sort_rules(rules):
sorted_rules = []
for rule in sorted(rules, key=lambda k: k.get('IPProtocol', '')):
if 'ports' in rule:
# Sort ports numerically, handle ranges through sorting by start port
rule['ports'] = sorted(rule['ports'], key=lambda k: int(k) if '-' not in k else int(k.split('-')[0]))
sorted_rules.append(rule)
return sorted_rules
| list: A list of string integers from number_1 to number_2. | ||
| """ | ||
| start, end = port_range.split('-') | ||
| return [str(i) for i in range(int(start), int(end) + 1)] |
There was a problem hiding this comment.
Nit: You could use xrange here so the whole list isn't built twice.
| ])) | ||
|
|
||
| def __lt__(self, other): | ||
| """Less than |
There was a problem hiding this comment.
Nit: First line of docstring should end with . (Same below)
| protocol = rule.get('IPProtocol') | ||
| if protocol == 'all': | ||
| self._applies_to_all = True | ||
| return self._applies_to_all |
There was a problem hiding this comment.
This return could be a break; then you only need the one return at the end.
| with self.assertRaises(ValueError): | ||
| _ = firewall_rule.FirewallAction(**action_1_dict) | ||
|
|
||
| @parameterized.parameterized.expand([ |
There was a problem hiding this comment.
I recommend having one or more parameterized tests where there are multiple protocols and ports listed in a rule. Include them in different orders in self and other to ensure sorting works correctly. Include ports in different orders as well.
| @@ -17,10 +17,14 @@ | |||
| See: https://cloud.google.com/compute/docs/reference/latest/firewalls | |||
| """ | |||
|
|
|||
| import netaddr | |||
There was a problem hiding this comment.
Your tests are failing because netaddr isn't included in the list of python modules to install in setup.py and scripts/docker/base
Codecov Report
@@ Coverage Diff @@
## dev #686 +/- ##
==========================================
+ Coverage 83.81% 83.95% +0.14%
==========================================
Files 165 165
Lines 8413 8515 +102
==========================================
+ Hits 7051 7149 +98
- Misses 1362 1366 +4
|
| self.destination_ranges == other.destination_ranges and | ||
| self.firewall_action == other.firewall_action) | ||
|
|
||
| def is_equivalent(self, other): |
There was a problem hiding this comment.
Add tests, please. You can see the code coverage and missed tests here: https://codecov.io/gh/GoogleCloudPlatform/forseti-security/pull/686/diff. There are a few other conditions below that are untested, add tests for them as well.
…seti-security into firewall-scanner
| """ | ||
| return (self.direction == other.direction and | ||
| self.network == other.network and | ||
| self.source_tags == other.source_tags and |
There was a problem hiding this comment.
Nit: This could use self._source_tags == other._source_tags and then it's an O(1) operation to compare two frozensets, which is just a hash value compare. Otherwise you are creating lists and comparing them.
Thanks for opening a Pull Request!
Here's a handy checklist to ensure your PR goes smoothly.
pylint --rcfile=pylintrcpasses.These guidelines and more can be found in our contributing guidelines.