gistfile1.txt
· 2.4 KiB · Text
Eredeti
def rule_resolve(rules_dir, rule_name, visited=None):
"""Recursively resolve a rule with all its base rules merged"""
if visited is None:
visited = set()
if rule_name in visited:
raise ValueError(f"Circular dependency detected: {rule_name}")
visited.add(rule_name)
rule_file = os.path.join(rules_dir, f"{rule_name}.rule")
with open(rule_file) as f:
rule = json.load(f)
# Start with empty merged rule
merged = {
'allow_ips': [],
'allow_ports': [],
'block_ips': [],
'block_ports': [],
'dns_redirect': False
}
# Merge base rules first (left to right)
for base_name in rule.get('extends', []):
base = rule_resolve(rules_dir, base_name, visited.copy())
merged['allow_ips'] += base.get('allow_ips', [])
merged['allow_ports'] += base.get('allow_ports', [])
merged['block_ips'] += base.get('block_ips', [])
merged['block_ports'] += base.get('block_ports', [])
if base.get('dns_redirect'):
merged['dns_redirect'] = True
# Child rule overrides — deduplicate
merged['allow_ips'] = list(dict.fromkeys(
merged['allow_ips'] + rule.get('allow_ips', [])))
merged['allow_ports'] = list(dict.fromkeys(
merged['allow_ports'] + rule.get('allow_ports', [])))
merged['block_ips'] = list(dict.fromkeys(
merged['block_ips'] + rule.get('block_ips', [])))
merged['block_ports'] = list(dict.fromkeys(
merged['block_ports'] + rule.get('block_ports', [])))
if rule.get('dns_redirect'):
merged['dns_redirect'] = True
merged['name'] = rule['name']
merged['desc'] = rule.get('desc', '')
merged['extends'] = rule.get('extends', [])
print(json.dumps(merged))
def find_rule_file(rules_dir, rule_name):
"""Find rule file in rules/ or rules/base/"""
for path in [
os.path.join(rules_dir, f"{rule_name}.rule"),
os.path.join(rules_dir, "base", f"{rule_name}.rule"),
]:
if os.path.exists(path):
return path
raise FileNotFoundError(f"Rule not found: {rule_name}")
'rule_resolve': lambda args: rule_resolve(args[0], args[1], args[2]),
'find_rule_file': lambda args: find_rule_file(args[0], args[1]),
Still need to add that to json.sh, also let's avoid inline Python, I saw you write it just now on rule::get()
| 1 | def rule_resolve(rules_dir, rule_name, visited=None): |
| 2 | """Recursively resolve a rule with all its base rules merged""" |
| 3 | if visited is None: |
| 4 | visited = set() |
| 5 | |
| 6 | if rule_name in visited: |
| 7 | raise ValueError(f"Circular dependency detected: {rule_name}") |
| 8 | visited.add(rule_name) |
| 9 | |
| 10 | rule_file = os.path.join(rules_dir, f"{rule_name}.rule") |
| 11 | with open(rule_file) as f: |
| 12 | rule = json.load(f) |
| 13 | |
| 14 | # Start with empty merged rule |
| 15 | merged = { |
| 16 | 'allow_ips': [], |
| 17 | 'allow_ports': [], |
| 18 | 'block_ips': [], |
| 19 | 'block_ports': [], |
| 20 | 'dns_redirect': False |
| 21 | } |
| 22 | |
| 23 | # Merge base rules first (left to right) |
| 24 | for base_name in rule.get('extends', []): |
| 25 | base = rule_resolve(rules_dir, base_name, visited.copy()) |
| 26 | merged['allow_ips'] += base.get('allow_ips', []) |
| 27 | merged['allow_ports'] += base.get('allow_ports', []) |
| 28 | merged['block_ips'] += base.get('block_ips', []) |
| 29 | merged['block_ports'] += base.get('block_ports', []) |
| 30 | if base.get('dns_redirect'): |
| 31 | merged['dns_redirect'] = True |
| 32 | |
| 33 | # Child rule overrides — deduplicate |
| 34 | merged['allow_ips'] = list(dict.fromkeys( |
| 35 | merged['allow_ips'] + rule.get('allow_ips', []))) |
| 36 | merged['allow_ports'] = list(dict.fromkeys( |
| 37 | merged['allow_ports'] + rule.get('allow_ports', []))) |
| 38 | merged['block_ips'] = list(dict.fromkeys( |
| 39 | merged['block_ips'] + rule.get('block_ips', []))) |
| 40 | merged['block_ports'] = list(dict.fromkeys( |
| 41 | merged['block_ports'] + rule.get('block_ports', []))) |
| 42 | if rule.get('dns_redirect'): |
| 43 | merged['dns_redirect'] = True |
| 44 | |
| 45 | merged['name'] = rule['name'] |
| 46 | merged['desc'] = rule.get('desc', '') |
| 47 | merged['extends'] = rule.get('extends', []) |
| 48 | |
| 49 | print(json.dumps(merged)) |
| 50 | |
| 51 | def find_rule_file(rules_dir, rule_name): |
| 52 | """Find rule file in rules/ or rules/base/""" |
| 53 | for path in [ |
| 54 | os.path.join(rules_dir, f"{rule_name}.rule"), |
| 55 | os.path.join(rules_dir, "base", f"{rule_name}.rule"), |
| 56 | ]: |
| 57 | if os.path.exists(path): |
| 58 | return path |
| 59 | raise FileNotFoundError(f"Rule not found: {rule_name}") |
| 60 | |
| 61 | |
| 62 | 'rule_resolve': lambda args: rule_resolve(args[0], args[1], args[2]), |
| 63 | 'find_rule_file': lambda args: find_rule_file(args[0], args[1]), |
| 64 | |
| 65 | |
| 66 | Still need to add that to json.sh, also let's avoid inline Python, I saw you write it just now on rule::get() |