id: dns-caa-issuer-policy-guard
namespace: company.team
description: Check domain DNS CAA records against a certificate-authority
allowlist. Fail closed on DNS errors.
inputs:
- id: domain
type: STRING
defaults: google.com
description: DNS name to check.
- id: root_domain
type: STRING
defaults: google.com
description: Stop the parent search at this domain. This must be a valid parent
of domain.
- id: allowed_issuers
type: STRING
defaults: pki.goog
description: Comma-separated issuer domain names that must match the effective
CAA policy.
- id: wildcard
type: BOOL
defaults: false
description: Set true to check wildcard issuance (issuewild).
- id: require_explicit
type: BOOL
defaults: true
description: Fail if the selected domain has no effective issuer policy.
triggers:
- id: weekday_policy_check
type: io.kestra.plugin.core.trigger.Schedule
cron: 0 7 * * 1-5
disabled: true
description: Disabled until the first manual check passes.
concurrency:
limit: 1
tasks:
- id: fetch_initial_caa
type: io.kestra.plugin.core.http.Request
description: Query Google DNS for the selected domain and expose the response in
task outputs.
uri: https://dns.google/resolve
method: GET
headers:
Accept: application/dns-json
params:
name: '{{ inputs.domain }}'
type: CAA
- id: check_caa
type: io.kestra.plugin.scripts.python.Script
description: Read the first DNS response, resolve parent policies when needed,
and write a CAA report.
containerImage: python:3.11-slim
env:
DOMAIN: '{{ inputs.domain }}'
ROOT_DOMAIN: '{{ inputs.root_domain }}'
ALLOWED_ISSUERS: '{{ inputs.allowed_issuers }}'
WILDCARD: '{{ inputs.wildcard }}'
REQUIRE_EXPLICIT: '{{ inputs.require_explicit }}'
inputFiles:
initial-caa-response.json: '{{ outputs.fetch_initial_caa.body | toJson }}'
outputFiles:
- caa-policy-report.json
script: |
"""Audit a domain's DNS CAA issuer policy. No Kestra-specific code in this module."""
import json
import os
import re
from urllib.parse import urlencode, urlsplit
from urllib.request import HTTPRedirectHandler, Request, build_opener
MAX_JSON_BYTES = 250_000
LABEL_RE = re.compile(r'^[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?$', re.I)
CAA_RE = re.compile(r'^([0-9]{1,3})\s+([a-zA-Z0-9]+)\s+("[^"\\]*"|\S+)$')
KNOWN_CAA_TAGS = {'issue', 'issuewild', 'iodef', 'contactemail', 'contactphone'}
def normalize_domain(value):
if not isinstance(value, str):
raise ValueError('Domain must be a string')
domain = value.strip().rstrip('.').lower()
if not 0 < len(domain) <= 253 or not all(LABEL_RE.fullmatch(x) for x in domain.split('.')):
raise ValueError(f'Invalid domain: {value!r}')
if len(domain.split('.')) < 2:
raise ValueError('Domain must have at least two labels')
return domain
def normalize_issuers(raw):
if not isinstance(raw, str):
raise ValueError('Issuer allowlist must be a string')
parts = [p.strip() for p in raw.split(',')]
if not parts or any(not p for p in parts):
raise ValueError('Issuer allowlist must contain nonempty domain names')
issuers = set(normalize_domain(p) for p in parts)
if len(issuers) != len(parts):
raise ValueError('Issuer allowlist contains duplicates')
return issuers
def names_to_query(domain, root_domain):
domain = normalize_domain(domain)
root = normalize_domain(root_domain)
if domain != root and not domain.endswith('.' + root):
raise ValueError('Domain is outside root_domain')
result = []
current = domain
while True:
result.append(current)
if current == root:
break
current = current.split('.', 1)[1]
return result
def parse_caa(data):
if not isinstance(data, str):
raise ValueError('CAA RDATA must be text')
match = CAA_RE.fullmatch(data.strip())
if not match:
raise ValueError(f'Unsupported CAA record encoding: {data[:120]}')
flags, tag, quoted = match.groups()
flags = int(flags)
if flags > 255:
raise ValueError('CAA flag exceeds 255')
tag = tag.lower()
value = quoted[1:-1] if quoted.startswith('"') else quoted
if not value and tag != 'iodef':
# Plain empty string is valid syntax for issue/issuewild and blocks issuance.
pass
if tag in ('issue', 'issuewild'):
issuer = value.split(';', 1)[0].strip().lower().rstrip('.')
if issuer:
issuer = normalize_domain(issuer)
else:
issuer = None
else:
issuer = None
return {'flags': flags, 'tag': tag, 'value': value, 'issuer': issuer}
class SameHostRedirect(HTTPRedirectHandler):
def redirect_request(self, request, fp, code, msg, headers, newurl):
dest = urlsplit(newurl)
if dest.scheme != 'https' or dest.hostname != 'dns.google' or dest.port not in (None, 443):
raise ValueError('DNS API redirect to an untrusted host')
return super().redirect_request(request, fp, code, msg, headers, newurl)
def google_doh(name):
url = 'https://dns.google/resolve?' + urlencode({'name': name, 'type': 'CAA'})
request = Request(url, headers={'Accept': 'application/dns-json', 'User-Agent': 'Kestra-CAA-Policy-Guard/1.0'})
with build_opener(SameHostRedirect()).open(request, timeout=15) as resp:
data = resp.read(MAX_JSON_BYTES + 1)
if len(data) > MAX_JSON_BYTES:
raise ValueError('DNS API response exceeds size limit')
return json.loads(data)
def caa_from_answer(payload, name):
if not isinstance(payload, dict) or type(payload.get('Status')) is not int:
raise ValueError(f'{name}: Invalid DNS JSON reply')
status = payload['Status']
if status != 0:
raise ValueError(f'{name}: DNS lookup returned status {status}; cannot audit')
if payload.get('TC') is True:
raise ValueError(f'{name}: Truncated DNS reply')
answers = payload.get('Answer', [])
if not isinstance(answers, list) or len(answers) > 100:
raise ValueError(f'{name}: Invalid DNS answer list')
out = []
for answer in answers:
if not isinstance(answer, dict) or type(answer.get('type')) is not int:
raise ValueError(f'{name}: Invalid DNS answer')
if answer['type'] == 5:
raise ValueError(f'{name}: CNAME needs separate resolution; refusing incomplete audit')
if answer['type'] == 257:
owner = normalize_domain(answer.get('name', ''))
if owner != name:
raise ValueError(f'{name}: Unexpected CAA owner {owner}')
out.append(parse_caa(answer.get('data')))
return out
def evaluate(records, allowed, wildcard, require_explicit):
unknown_critical = [r['tag'] for r in records if (r['flags'] & 128) and r['tag'] not in KNOWN_CAA_TAGS]
effective = [r for r in records if r['tag'] == ('issuewild' if wildcard and any(x['tag'] == 'issuewild' for x in records) else 'issue')]
declared = {r['issuer'] for r in effective if r['issuer']}
denies_all = bool(effective) and not declared
failures = []
if unknown_critical:
failures.append('Unknown critical CAA tag: ' + ', '.join(sorted(set(unknown_critical))))
if not effective and require_explicit:
failures.append('No explicit issuer policy for this certificate type')
if denies_all:
failures.append('CAA policy denies all issuers')
extra = sorted(declared - allowed)
missing = sorted(allowed - declared) if effective else []
if extra:
failures.append('Unapproved issuers: ' + ', '.join(extra))
if missing:
failures.append('Expected issuers not declared: ' + ', '.join(missing))
return {
'status': 'FAIL' if failures else 'PASS',
'effective_tag': 'issuewild' if wildcard and any(x['tag'] == 'issuewild' for x in records) else 'issue',
'effective_issuers': sorted(declared),
'failures': failures,
}
def audit(domain, root_domain, allowlist, wildcard=False, require_explicit=True, lookup=google_doh):
chain = names_to_query(domain, root_domain)
allowed = normalize_issuers(allowlist)
if not isinstance(wildcard, bool) or not isinstance(require_explicit, bool):
raise ValueError('Boolean settings are required')
checked = []
records = []
source = None
for name in chain:
checked.append(name)
records = caa_from_answer(lookup(name), name)
if records:
source = name
break
outcome = evaluate(records, allowed, wildcard, require_explicit)
return {
'domain': chain[0],
'root_domain': chain[-1],
'policy_source': source,
'checked_names': checked,
'certificate_type': 'WILDCARD' if wildcard else 'NORMAL',
'allowed_issuers': sorted(allowed),
'records': records,
**outcome,
}
def bool_env(name, default):
raw = os.getenv(name, default).strip().lower()
if raw not in ('true', 'false'):
raise ValueError(f'{name} must be true or false')
return raw == 'true'
def read_initial_response(filename='initial-caa-response.json'):
# Kestra may render an HTTP body as an object or as a JSON string.
with open(filename, encoding='utf-8') as f:
response = json.load(f)
if isinstance(response, str):
response = json.loads(response)
if not isinstance(response, dict):
raise ValueError('Initial CAA reply must be a JSON object')
return response
def main():
domain = normalize_domain(os.environ['DOMAIN'])
initial = read_initial_response()
def lookup(name):
return initial if name == domain else google_doh(name)
report = audit(
domain, os.environ['ROOT_DOMAIN'], os.environ['ALLOWED_ISSUERS'],
wildcard=bool_env('WILDCARD', 'false'),
require_explicit=bool_env('REQUIRE_EXPLICIT', 'true'),
lookup=lookup,
)
with open('caa-policy-report.json', 'w', encoding='utf-8') as f:
json.dump(report, f, indent=2, sort_keys=True)
print('::' + json.dumps({'outputs': {
'status': report['status'],
'policy_source': report['policy_source'] or 'none',
'issuer_count': len(report['effective_issuers']),
'failures': '; '.join(report['failures']) or 'none',
}}, separators=(',', ':')) + '::')
print(f"CAA audit: {report['domain']} => {report['status']} ({report['policy_source'] or 'no CAA'})")
if __name__ == '__main__':
main()
- id: policy_result
type: io.kestra.plugin.core.flow.If
description: Send an alert when the policy does not match the allowlist.
condition: '{{ outputs.check_caa.vars.status == ''FAIL'' }}'
then:
- id: alert_caa_violation
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
description: Alert on an invalid issuer policy.
url: '{{ secret(''SLACK_WEBHOOK_URL'') }}'
payload: '{"text": "CAA policy FAIL for {{ inputs.domain }}. Source: {{
outputs.check_caa.vars.policy_source }}. {{
outputs.check_caa.vars.failures }}. Execution: {{ execution.id }}"}'
- id: log_failed_policy
type: io.kestra.plugin.core.log.Log
message: 'CAA policy mismatch for {{ inputs.domain }}: {{
outputs.check_caa.vars.failures }}.'
else:
- id: log_passed_policy
type: io.kestra.plugin.core.log.Log
message: CAA policy passed for {{ inputs.domain }}; source {{
outputs.check_caa.vars.policy_source }}.
- id: report_location
type: io.kestra.plugin.core.output.OutputValues
description: Publish a report URI and the check status.
values:
status: '{{ outputs.check_caa.vars.status }}'
report_uri: '{{ outputs.check_caa.outputFiles[''caa-policy-report.json''] }}'
policy_source: '{{ outputs.check_caa.vars.policy_source }}'
errors:
- id: log_check_failure
type: io.kestra.plugin.core.log.Log
message: 'CAA audit failed before it could confirm policy compliance. Execution:
{{ execution.id }}.'
