source

@traced
async def filter_rulebase(
    self,
    mgmt_name: str,
    domain_name: str | None = None,
    *,
    package: str | None = None,
    layer: str | None = None,
    rulebase_type: RulebaseType = "access",
    filter_query: str = "",
    output_format: RuleFormat = "model_friendly",
    cache_mode: CacheMode = "cache",
    cache_ttl: float | None = None,
) -> FilteredRulebaseResult:
    """Evaluate `filter_query` locally against cached rulebase and cached objects/services."""
    if not package and not layer:
        raise ValueError("Either 'package' or 'layer' must be specified for filter_rulebase.")

    if package is not None:
        resolved_domain = await self._resolve_domain_for_target(
            mgmt_name,
            domain_name,
            package,
            kind="package",
            rulebase_type=rulebase_type,
        )
        snap, session_uid, published_at, status, age = await self._ensure_and_load_snapshot(
            mgmt_name,
            resolved_domain,
            cache_mode=cache_mode,
            cache_ttl=cache_ttl,
        )
        rb: PackageRulebase | NumberedLayerRulebase = number_package_rulebase(
            snap,
            package,
            rulebase_type=rulebase_type,
            snapshot_session_uid=session_uid,
            snapshot_published_at=published_at,
            sync_status=status,
        )
    else:
        assert layer is not None
        resolved_domain = await self._resolve_domain_for_target(
            mgmt_name,
            domain_name,
            layer,
            kind="layer",
            rulebase_type=rulebase_type,
        )
        snap, session_uid, published_at, status, age = await self._ensure_and_load_snapshot(
            mgmt_name,
            resolved_domain,
            cache_mode=cache_mode,
            cache_ttl=cache_ttl,
        )
        rb = number_layer_rulebase(
            snap,
            layer,
            rulebase_type=rulebase_type,
            prefix="",
            expand_placeholders=False,
            expand_inline_layers=True,
            snapshot_session_uid=session_uid,
            snapshot_published_at=published_at,
            sync_status=status,
        )

    # Load cached objects and services for rich IP/CIDR/group/port filter matching
    if self._object_service is not None:
        objects = await self._object_service.get_objects(
            mgmt_name=mgmt_name,
            domain=resolved_domain,
            include_global=True,
            cache_mode="cache",
        )
    else:
        objects = await self._cache.get_objects(mgmt_name=mgmt_name, domain_name=resolved_domain)

    if self._service_cache is not None:
        services = await self._service_cache.get_services(
            mgmt_name=mgmt_name,
            domain=resolved_domain,
            include_global=True,
            cache_mode="cache",
        )
    else:
        services = await self._cache.get_services(mgmt_name=mgmt_name, domain_name=resolved_domain)

    return filter_numbered_rulebase(
        rb,
        filter_query=filter_query,
        output_format=output_format,
        objects=objects,
        services=services,
        cache_age_seconds=age,
    )

@traced
async def get_layer_rulebase(
    self,
    mgmt_name: str,
    domain_name: str | None,
    layer: str,
    *,
    rulebase_type: RulebaseType | None = None,
    cache_mode: CacheMode = "cache",
    cache_ttl: float | None = None,
) -> NumberedLayerRulebase:
    """Return a SmartConsole-numbered `NumberedLayerRulebase` for `layer` in `(mgmt_name, domain_name)`."""
    resolved_domain = await self._resolve_domain_for_target(
        mgmt_name,
        domain_name,
        layer,
        kind="layer",
        rulebase_type=rulebase_type,
    )
    span_attrs(
        mgmt_name=mgmt_name,
        domain=resolved_domain,
        layer=layer,
        rulebase_type=rulebase_type,
        cache_mode=cache_mode,
    )
    snap, session_uid, published_at, status, _ = await self._ensure_and_load_snapshot(
        mgmt_name,
        resolved_domain,
        cache_mode=cache_mode,
        cache_ttl=cache_ttl,
    )
    return number_layer_rulebase(
        snap,
        layer,
        rulebase_type=rulebase_type,
        prefix="",
        expand_placeholders=False,
        expand_inline_layers=True,
        snapshot_session_uid=session_uid,
        snapshot_published_at=published_at,
        sync_status=status,
    )

@traced
async def get_package_rulebase(
    self,
    mgmt_name: str,
    domain_name: str | None,
    package: str,
    *,
    rulebase_type: RulebaseType = "access",
    cache_mode: CacheMode = "cache",
    cache_ttl: float | None = None,
) -> PackageRulebase:
    """Return a SmartConsole-numbered `PackageRulebase` for `package` in `(mgmt_name, domain_name)`."""
    resolved_domain = await self._resolve_domain_for_target(
        mgmt_name,
        domain_name,
        package,
        kind="package",
        rulebase_type=rulebase_type,
    )
    span_attrs(
        mgmt_name=mgmt_name,
        domain=resolved_domain,
        package=package,
        rulebase_type=rulebase_type,
        cache_mode=cache_mode,
    )
    snap, session_uid, published_at, status, _ = await self._ensure_and_load_snapshot(
        mgmt_name,
        resolved_domain,
        cache_mode=cache_mode,
        cache_ttl=cache_ttl,
    )
    return number_package_rulebase(
        snap,
        package,
        rulebase_type=rulebase_type,
        snapshot_session_uid=session_uid,
        snapshot_published_at=published_at,
        sync_status=status,
    )

@traced
async def get_policy_packages(
    self,
    mgmt_name: str,
    domain_name: str | None = None,
    *,
    cache_mode: CacheMode = "cache",
    cache_ttl: float | None = None,
) -> list[PolicyPackageInfo]:
    """Return cached `PolicyPackageInfo` entries for `(mgmt_name, domain_name)`."""
    span_attrs(mgmt_name=mgmt_name, domain=domain_name, cache_mode=cache_mode)
    if domain_name is None or not domain_name.strip():
        domains = await self._list_candidate_domains(mgmt_name)
        non_global = [d for d in domains if d != "Global"]
        target_domains = non_global or domains
        results: list[PolicyPackageInfo] = []
        for dom in target_domains:
            snap, _, _, _, _ = await self._ensure_and_load_snapshot(
                mgmt_name,
                dom,
                cache_mode=cache_mode,
                cache_ttl=cache_ttl,
            )
            results.extend(extract_policy_packages(snap))
        return results

    snap, _, _, _, _ = await self._ensure_and_load_snapshot(
        mgmt_name,
        domain_name.strip(),
        cache_mode=cache_mode,
        cache_ttl=cache_ttl,
    )
    return extract_policy_packages(snap)

@traced
async def locate_rules(
    self,
    mgmt_name: str,
    domain_name: str | None = None,
    *,
    rule_uids: Sequence[str] | None = None,
    layer_uids: Sequence[str] | None = None,
    cache_mode: CacheMode = "cache",
    cache_ttl: float | None = None,
) -> RuleLocations:
    """Locate SmartConsole rule numbers and layer prefixes across all packages and layers of a domain."""
    if domain_name is None or not domain_name.strip():
        domains = await self._list_candidate_domains(mgmt_name)
        non_global = [d for d in domains if d != "Global"]
        if len(non_global) == 1:
            resolved_domain = non_global[0]
        elif len(domains) == 1:
            resolved_domain = domains[0]
        else:
            raise DomainNotFoundError(
                f"Domain name is required for locate_rules on multi-domain server {mgmt_name!r}."
            )
    else:
        resolved_domain = domain_name.strip()

    span_attrs(
        mgmt_name=mgmt_name,
        domain=resolved_domain,
        rule_count=len(rule_uids) if rule_uids else 0,
        layer_count=len(layer_uids) if layer_uids else 0,
    )
    snap, session_uid, published_at, status, _ = await self._ensure_and_load_snapshot(
        mgmt_name,
        resolved_domain,
        cache_mode=cache_mode,
        cache_ttl=cache_ttl,
    )
    return locate_rules_in_snapshot(
        snap,
        rule_uids=rule_uids,
        layer_uids=layer_uids,
        snapshot_session_uid=session_uid,
        snapshot_published_at=published_at,
        sync_status=status,
    )