@traced
async def fetch_paged_rulebase(
api: ApiPort,
mgmt_name: str,
domain_name: str | None,
*,
rulebase_type: RulebaseType,
layer_uid: str = "",
layer_name: str = "",
package_uid: str | None = None,
package_name: str | None = None,
details_level: str = "standard",
use_object_dictionary: bool = True,
limit: int = MAX_RULEBASE_PAGE_LIMIT,
extra_payload: dict[str, Any] | None = None,
sid: str | None = None,
server_ip: str | None = None,
) -> PagedRulebaseResult:
"""Page `show-*-rulebase` in chunks of up to 100 rules, stitching split sections and preserving empty trailing sections."""
page_limit = max(1, min(int(limit), MAX_RULEBASE_PAGE_LIMIT))
command = RULEBASE_TYPE_TO_COMMAND[rulebase_type]
eff_domain = domain_name or ""
target_layer_uid = layer_uid
target_layer_name = layer_name
if rulebase_type == "nat":
target_layer_uid = target_layer_uid or (package_uid or package_name or "")
target_layer_name = target_layer_name or (package_name or package_uid or "")
span_attrs(
mgmt_name=mgmt_name,
domain=domain_name,
rulebase_type=rulebase_type,
layer_uid=target_layer_uid or None,
layer_name=target_layer_name or None,
package_name=package_name,
limit=page_limit,
)
combined_obj_dict: dict[str, dict[str, Any]] = {}
sections_by_key: dict[str, RulebaseSectionEntry] = {}
section_order: list[str] = []
raw_sections_by_key: dict[str, dict[str, Any]] = {}
seen_rule_uids: set[str] = set()
ordered_raw_rules: list[tuple[dict[str, Any], str | None, str | None]] = []
offset = 0
total_reported: int | None = None
resolved_layer_uid = target_layer_uid
resolved_layer_name = target_layer_name
did_trailing_probe = False
while True:
payload: dict[str, Any] = {
"limit": page_limit,
"offset": offset,
"details-level": details_level,
"use-object-dictionary": use_object_dictionary,
}
if rulebase_type == "nat":
pkg_identifier = package_name or package_uid or target_layer_name or target_layer_uid
if pkg_identifier:
payload["package"] = pkg_identifier
else:
if target_layer_uid:
payload["uid"] = target_layer_uid
elif target_layer_name:
payload["name"] = target_layer_name
if package_name:
payload["package"] = package_name
elif package_uid:
payload["package"] = package_uid
if extra_payload:
payload.update(extra_payload)
res = await api.api_call(
mgmt_name,
command,
payload,
domain=domain_name,
sid=sid,
server_ip=server_ip,
)
if not res.success:
raise RuntimeError(
f"Failed to fetch {command} for {target_layer_name or target_layer_uid or package_name!r} "
f"on {mgmt_name}:{eff_domain}: {res.error_message or res.error_code or 'API error'}"
)
data = res.data if isinstance(res.data, dict) else {}
if not resolved_layer_uid and isinstance(data.get("uid"), str) and data["uid"]:
resolved_layer_uid = str(data["uid"])
if not resolved_layer_name and isinstance(data.get("name"), str) and data["name"]:
resolved_layer_name = str(data["name"])
# Collect objects-dictionary entries from this page
raw_obj_dict = data.get("objects-dictionary")
if isinstance(raw_obj_dict, list):
for obj_item in raw_obj_dict:
if isinstance(obj_item, dict) and isinstance(obj_item.get("uid"), str):
combined_obj_dict[obj_item["uid"]] = obj_item
elif isinstance(raw_obj_dict, dict):
for k, v in raw_obj_dict.items():
if isinstance(v, dict):
combined_obj_dict[str(k)] = v
raw_rulebase = data.get("rulebase")
nodes = raw_rulebase if isinstance(raw_rulebase, list) else []
if total_reported is None and isinstance(data.get("total"), int):
total_reported = int(data["total"])
rules_seen_in_page = 0
for node in nodes:
if not isinstance(node, dict):
continue
if _is_section_node(node):
sec_uid = str(node.get("uid") or node.get("name") or f"sec:{len(section_order) + 1}")
sec_name = str(node.get("name") or "")
from_val = node.get("from")
to_val = node.get("to")
sec_from = int(from_val) if isinstance(from_val, int) and from_val > 0 else None
sec_to = int(to_val) if isinstance(to_val, int) and to_val > 0 else None
nested_items = node.get("rulebase")
child_rules = nested_items if isinstance(nested_items, list) else []
if sec_uid not in sections_by_key:
sec_pos = len(section_order) + 1
eff_l_uid = resolved_layer_uid or target_layer_uid or (package_uid or "")
eff_l_name = resolved_layer_name or target_layer_name or (package_name or "")
sec_entry = RulebaseSectionEntry(
mgmt_name=mgmt_name,
domain_name=eff_domain,
layer_uid=eff_l_uid,
layer_name=eff_l_name,
rulebase_type=rulebase_type,
uid=sec_uid,
name=sec_name,
position=sec_pos,
from_rule=sec_from,
to_rule=sec_to,
rule_uids=[],
raw_data=dict(node),
updated_at=utc_now(),
)
sections_by_key[sec_uid] = sec_entry
section_order.append(sec_uid)
raw_sections_by_key[sec_uid] = dict(node)
raw_sections_by_key[sec_uid]["rulebase"] = []
else:
sec_entry = sections_by_key[sec_uid]
if sec_entry.from_rule is None and sec_from is not None:
sec_entry.from_rule = sec_from
if sec_to is not None:
sec_entry.to_rule = sec_to
raw_sections_by_key[sec_uid]["to"] = sec_to
for child in child_rules:
if not isinstance(child, dict):
continue
rules_seen_in_page += 1
r_uid = str(child.get("uid") or f"rule:{len(ordered_raw_rules) + 1}")
if r_uid not in sec_entry.rule_uids:
sec_entry.rule_uids.append(r_uid)
raw_sections_by_key[sec_uid]["rulebase"].append(child)
if r_uid not in seen_rule_uids:
seen_rule_uids.add(r_uid)
ordered_raw_rules.append((child, sec_uid, sec_name))
else:
rules_seen_in_page += 1
r_uid = str(node.get("uid") or f"rule:{len(ordered_raw_rules) + 1}")
if r_uid not in seen_rule_uids:
seen_rule_uids.add(r_uid)
ordered_raw_rules.append((node, None, None))
to_index = data.get("to")
if isinstance(to_index, int) and to_index > offset:
next_offset = to_index
elif rules_seen_in_page > 0:
next_offset = offset + rules_seen_in_page
else:
break
if total_reported is not None:
if next_offset < total_reported and rules_seen_in_page > 0:
offset = next_offset
continue
# If we just reached total_reported on a full page boundary (e.g. 100 of 100),
# do a single trailing probe at offset=total_reported in case Check Point has
# an empty trailing section after the last rule.
if (
not did_trailing_probe
and total_reported > 0
and next_offset == total_reported
and rules_seen_in_page >= page_limit
):
did_trailing_probe = True
offset = next_offset
continue
break
if rules_seen_in_page < page_limit:
break
offset = next_offset
final_layer_uid = resolved_layer_uid or target_layer_uid or (package_uid or "")
final_layer_name = resolved_layer_name or target_layer_name or (package_name or "")
final_pkg_uid = package_uid or final_layer_uid
final_pkg_name = package_name or final_layer_name
# Finalize section entries with resolved layer_uid/layer_name and from_rule/to_rule bounds
rule_pos_by_uid: dict[str, int] = {}
access_rules: list[AccessRuleEntry] = []
nat_rules: list[NatRuleEntry] = []
https_rules: list[HttpsRuleEntry] = []
threat_rules: list[ThreatRuleEntry] = []
raw_rules_list: list[dict[str, Any]] = []
for idx, (raw_rule, rule_sec_uid, rule_sec_name) in enumerate(ordered_raw_rules, start=1):
raw_rules_list.append(raw_rule)
r_uid = str(raw_rule.get("uid") or f"rule:{idx}")
rule_pos_by_uid[r_uid] = idx
if rulebase_type == "access":
access_rules.append(
parse_access_rule_entry(
mgmt_name=mgmt_name,
domain_name=eff_domain,
layer_uid=final_layer_uid,
layer_name=final_layer_name,
raw_rule=raw_rule,
position=idx,
section_uid=rule_sec_uid,
section_name=rule_sec_name,
obj_dict=combined_obj_dict,
)
)
elif rulebase_type == "nat":
nat_rules.append(
parse_nat_rule_entry(
mgmt_name=mgmt_name,
domain_name=eff_domain,
package_uid=final_pkg_uid,
package_name=final_pkg_name,
raw_rule=raw_rule,
position=idx,
section_uid=rule_sec_uid,
section_name=rule_sec_name,
obj_dict=combined_obj_dict,
)
)
elif rulebase_type == "https":
https_rules.append(
parse_https_rule_entry(
mgmt_name=mgmt_name,
domain_name=eff_domain,
layer_uid=final_layer_uid,
layer_name=final_layer_name,
raw_rule=raw_rule,
position=idx,
section_uid=rule_sec_uid,
section_name=rule_sec_name,
obj_dict=combined_obj_dict,
)
)
elif rulebase_type == "threat":
threat_rules.append(
parse_threat_rule_entry(
mgmt_name=mgmt_name,
domain_name=eff_domain,
layer_uid=final_layer_uid,
layer_name=final_layer_name,
raw_rule=raw_rule,
position=idx,
section_uid=rule_sec_uid,
section_name=rule_sec_name,
obj_dict=combined_obj_dict,
)
)
final_sections: list[RulebaseSectionEntry] = []
final_raw_sections: list[dict[str, Any]] = []
for pos, s_key in enumerate(section_order, start=1):
sec = sections_by_key[s_key]
positions = [rule_pos_by_uid[u] for u in sec.rule_uids if u in rule_pos_by_uid]
from_pos = min(positions) if positions else sec.from_rule
to_pos = max(positions) if positions else sec.to_rule
if not sec.rule_uids:
from_pos = None
to_pos = None
updated_sec = RulebaseSectionEntry(
mgmt_name=mgmt_name,
domain_name=eff_domain,
layer_uid=final_layer_uid,
layer_name=final_layer_name,
rulebase_type=rulebase_type,
uid=sec.uid,
name=sec.name,
position=pos,
from_rule=from_pos,
to_rule=to_pos,
rule_uids=list(sec.rule_uids),
raw_data=dict(raw_sections_by_key.get(s_key, sec.raw_data)),
updated_at=utc_now(),
)
final_sections.append(updated_sec)
final_raw_sections.append(raw_sections_by_key[s_key])
return PagedRulebaseResult(
mgmt_name=mgmt_name,
domain_name=eff_domain,
rulebase_type=rulebase_type,
layer_uid=final_layer_uid,
layer_name=final_layer_name,
package_uid=package_uid,
package_name=package_name,
sections=final_sections,
access_rules=access_rules,
nat_rules=nat_rules,
https_rules=https_rules,
threat_rules=threat_rules,
objects_dictionary=combined_obj_dict,
raw_rules=raw_rules_list,
raw_sections=final_raw_sections,
total=len(ordered_raw_rules),
)