gitea-admin/gh-christianlouis-dmarq
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix DKIM selector extraction: show all working selectors and report-discovered selectors

Browse Source 
- check_dkim now returns ALL matching selectors instead of stopping at first match
- DomainDNSResult.dkim_selectors is now a List[str] instead of a single Optional[str]
- DNSRecordResponse.dkimSelectors is now List[str]
- /selectors endpoint now also returns report_selectors (auto-discovered from DMARC reports)
- Frontend shows all live-check selectors and auto-discovered selectors as read-only
- Updated tests to match new data structures; added tests for multi-selector and report_selectors

Agent-Logs-Url: https://github.com/christianlouis/dmarq/sessions/87d8b8d9-23c3-4d3b-85a3-8e354e62c768

Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]
2026-03-29 23:54:45 +00:00
parent 985f49d61f
commit 596a9b882e
5 changed files with 121 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/api/api_v1/endpoints/domains.py
+13 -5
View File
@@ -55,7 +55,7 @@ class DNSRecordResponse(BaseModel):
spf: bool
spfRecord: Optional[str] = None
dkim: bool
dkimSelectors: Optional[str] = None
dkimSelectors: List[str] = []
class TimelinePoint(BaseModel):
@@ -347,7 +347,7 @@ async def get_domain_dns_records(
spf=result.spf,
spfRecord=result.spf_record,
dkim=result.dkim,
dkimSelectors=result.dkim_selector,
dkimSelectors=result.dkim_selectors,
)
@@ -514,15 +514,23 @@ async def get_domain_selectors(
domain_id: str = Path(..., title="The domain ID or name"),
db: Session = Depends(get_db),
):
"""Return the manually configured DKIM selectors for a domain."""
"""Return the manually configured DKIM selectors for a domain.
The response includes both ``selectors`` (manually configured, can be
deleted) and ``report_selectors`` (automatically discovered from received
DMARC reports, read-only).
"""
store = ReportStore.get_instance()
if domain_id not in store.get_domains():
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Domain not found",
)
selectors = _get_domain_selectors_from_db(db, domain_id)
return {"selectors": selectors}
manual = _get_domain_selectors_from_db(db, domain_id)
report = _get_selectors_from_reports(store, domain_id)
# Only include in report_selectors those not already in the manual list
auto = [s for s in report if s not in manual]
return {"selectors": manual, "report_selectors": auto}
@router.post("/{domain_id}/selectors", status_code=status.HTTP_201_CREATED)
backend/app/services/dns_resolver.py
+19 -7
View File
@@ -71,7 +71,8 @@ class DomainDNSResult:
spf: bool = False
spf_record: Optional[str] = None
dkim: bool = False
dkim_selector: Optional[str] = None
# All selectors that resolved to a valid DKIM record (may be multiple)
dkim_selectors: List[str] = field(default_factory=list)
dkim_record: Optional[str] = None
# Track which selectors were tried so callers can surface this information
selectors_checked: List[str] = field(default_factory=list)
@@ -133,14 +134,25 @@ class BaseDNSProvider(ABC):
async def check_dkim(
self, domain: str, selectors: List[str]
) -> Tuple[bool, Optional[str], Optional[str]]:
"""Return *(found, selector, record_string)* for the first working DKIM selector."""
) -> Tuple[bool, List[str], Optional[str]]:
"""Return *(found, matching_selectors, first_record_string)* for all working DKIM selectors.
All selectors in *selectors* are checked and every one that resolves to
a valid DKIM TXT record is collected. The boolean is ``True`` when at
least one selector resolved. *first_record_string* is the record text
for the first matching selector (useful for display purposes).
"""
matching_selectors: List[str] = []
first_record: Optional[str] = None
for selector in selectors:
try:
records = await self.lookup_txt(f"{selector}._domainkey.{domain}")
for record in records:
if "v=dkim1" in record.lower() or "p=" in record.lower():
return True, selector, record
matching_selectors.append(selector)
if first_record is None:
first_record = record
break
except LookupError as exc:
logger.debug(
"DKIM lookup failed for selector=%s domain=%s: %s",
@@ -148,7 +160,7 @@ class BaseDNSProvider(ABC):
_sanitize_for_log(domain),
exc,
)
return False, None, None
return bool(matching_selectors), matching_selectors, first_record
async def check_domain(
self, domain: str, selectors: Optional[List[str]] = None
@@ -169,7 +181,7 @@ class BaseDNSProvider(ABC):
spf_coro = self.check_spf(domain)
dkim_coro = self.check_dkim(domain, all_selectors)
(dmarc_ok, dmarc_record), (spf_ok, spf_record), (dkim_ok, dkim_sel, dkim_record) = (
(dmarc_ok, dmarc_record), (spf_ok, spf_record), (dkim_ok, dkim_sels, dkim_record) = (
await asyncio.gather(dmarc_coro, spf_coro, dkim_coro)
)
@@ -179,7 +191,7 @@ class BaseDNSProvider(ABC):
spf=spf_ok,
spf_record=spf_record,
dkim=dkim_ok,
dkim_selector=dkim_sel,
dkim_selectors=dkim_sels,
dkim_record=dkim_record,
selectors_checked=all_selectors,
)
backend/app/templates/domain_details.html
+23 -9
View File
@@ -169,7 +169,7 @@
Manually configure selectors to check. Selectors seen in received DMARC
reports and common well-known selectors are checked automatically.
</p>
<!-- Existing selectors list -->
<!-- Manually configured selectors -->
<div class="mb-3">
<template x-if="selectors.length === 0">
<p class="text-sm text-muted-foreground italic">No manually configured selectors yet.</p>
@@ -185,6 +185,18 @@
</div>
</template>
</div>
<!-- Auto-discovered selectors from DMARC reports -->
<template x-if="reportSelectors.length > 0">
<div class="mb-3">
<p class="text-xs text-muted-foreground font-semibold uppercase tracking-wide mb-1">From DMARC reports (auto-detected)</p>
<template x-for="sel in reportSelectors" :key="sel">
<div class="flex items-center py-1 px-2 rounded bg-muted/50 border border-dashed mb-1">
<span class="font-mono text-sm text-muted-foreground" x-text="sel"></span>
<span class="ml-2 text-xs text-muted-foreground italic">auto</span>
</div>
</template>
</div>
</template>
<!-- Add selector form -->
<div class="flex items-center gap-2">
<input
@@ -445,9 +457,10 @@ function domainDetailsApp(domainId) {
spf: false,
spfRecord: '',
dkim: false,
dkimSelectors: ''
dkimSelectors: []
},
selectors: [],
reportSelectors: [],
newSelector: '',
selectorError: '',
reports: [],
@@ -481,7 +494,9 @@ function domainDetailsApp(domainId) {
get dkimLiveText() {
if (!this.dns.dkim) return 'No DKIM record found for configured selectors';
if (this.dns.dkimSelectors) return 'selector: ' + this.dns.dkimSelectors;
if (this.dns.dkimSelectors && this.dns.dkimSelectors.length > 0) {
return 'selectors: ' + this.dns.dkimSelectors.join(', ');
}
return 'Verified';
},
@@ -515,6 +530,7 @@ function domainDetailsApp(domainId) {
if (response.ok) {
const data = await response.json();
this.selectors = data.selectors || [];
this.reportSelectors = data.report_selectors || [];
}
} catch (error) {
console.error('Error fetching selectors:', error);
@@ -532,10 +548,9 @@ function domainDetailsApp(domainId) {
body: JSON.stringify({ selector: sel })
});
if (response.ok) {
const data = await response.json();
this.selectors = data.selectors || [];
this.newSelector = '';
// Refresh DNS check to reflect the new selector
// Refresh selectors (both manual and report) and DNS check
this.fetchSelectors();
this.fetchDNSRecords();
} else {
const err = await response.json();
@@ -554,9 +569,8 @@ function domainDetailsApp(domainId) {
{ method: 'DELETE' }
);
if (response.ok) {
const data = await response.json();
this.selectors = data.selectors || [];
// Refresh DNS check after removing a selector
// Refresh selectors (both manual and report) and DNS check
this.fetchSelectors();
this.fetchDNSRecords();
} else {
console.error('Error deleting selector:', response.status);
backend/app/tests/test_dns_endpoints.py
+40 -2
View File
@@ -49,7 +49,7 @@ MOCK_DNS_RESULT = DomainDNSResult(
spf=True,
spf_record="v=spf1 include:_spf.google.com ~all",
dkim=True,
dkim_selector="google",
dkim_selectors=["google"],
dkim_record="v=DKIM1; k=rsa; p=ABC",
)
@@ -79,7 +79,9 @@ def test_get_selectors_empty(client: TestClient):
"""Returns an empty list when no selectors have been configured."""
response = client.get(f"/api/v1/domains/{DOMAIN}/selectors")
assert response.status_code == 200
assert response.json() == {"selectors": []}
data = response.json()
assert data["selectors"] == []
assert "report_selectors" in data
def test_get_selectors_unknown_domain(client: TestClient):
@@ -176,6 +178,42 @@ def test_delete_selector_unknown_domain(client: TestClient):
# ---------------------------------------------------------------------------
def test_get_selectors_includes_report_selectors(client: TestClient):
"""Report selectors (from DMARC report records) are returned in report_selectors."""
response = client.get(f"/api/v1/domains/{DOMAIN}/selectors")
assert response.status_code == 200
data = response.json()
# The MINIMAL_REPORT has a record with selector "google" in its dkim auth results
assert "google" in data["report_selectors"]
def test_get_selectors_report_selector_moves_to_manual_when_added(client: TestClient):
"""A selector discovered from reports should appear only in 'selectors' once added manually."""
# Confirm it's in report_selectors before adding
r1 = client.get(f"/api/v1/domains/{DOMAIN}/selectors")
assert "google" in r1.json()["report_selectors"]
# Add it as a manual selector
client.post(f"/api/v1/domains/{DOMAIN}/selectors", json={"selector": "google"})
r2 = client.get(f"/api/v1/domains/{DOMAIN}/selectors")
data = r2.json()
assert "google" in data["selectors"]
# It must not appear in both lists
assert "google" not in data["report_selectors"]
def test_dns_endpoint_returns_dkim_selectors_as_list(client: TestClient):
"""The /dns endpoint should return dkimSelectors as a list."""
with _mock_dns():
response = client.get(f"/api/v1/domains/{DOMAIN}/dns")
assert response.status_code == 200
data = response.json()
assert isinstance(data["dkimSelectors"], list)
assert "google" in data["dkimSelectors"]
def test_dns_endpoint_returns_real_data(client: TestClient):
"""The /dns endpoint should return the mocked DNS check result."""
with _mock_dns():
backend/app/tests/test_dns_resolver.py
+26 -9
View File
@@ -110,9 +110,9 @@ async def test_check_dkim_found_first_selector():
provider = FakeDNSProvider(
{"google._domainkey.example.com": ["v=DKIM1; k=rsa; p=MIGfMA0GCSqGSIb3"]}
)
found, selector, record = await provider.check_dkim("example.com", ["google", "mail"])
found, selectors, record = await provider.check_dkim("example.com", ["google", "mail"])
assert found is True
assert selector == "google"
assert selectors == ["google"]
assert record is not None
@@ -121,17 +121,34 @@ async def test_check_dkim_found_second_selector():
provider = FakeDNSProvider(
{"mail._domainkey.example.com": ["v=DKIM1; k=rsa; p=MIGfMA0GCSqGSIb3"]}
)
found, selector, record = await provider.check_dkim("example.com", ["google", "mail"])
found, selectors, record = await provider.check_dkim("example.com", ["google", "mail"])
assert found is True
assert selector == "mail"
assert selectors == ["mail"]
@pytest.mark.asyncio
async def test_check_dkim_found_multiple_selectors():
"""When multiple selectors resolve, all are returned."""
provider = FakeDNSProvider(
{
"google._domainkey.example.com": ["v=DKIM1; k=rsa; p=MIGfMA0GCSqGSIb3"],
"mail._domainkey.example.com": ["v=DKIM1; k=rsa; p=XYZ"],
}
)
found, selectors, record = await provider.check_dkim("example.com", ["google", "mail"])
assert found is True
assert "google" in selectors
assert "mail" in selectors
assert len(selectors) == 2
assert record is not None # record of the first match
@pytest.mark.asyncio
async def test_check_dkim_not_found():
provider = FakeDNSProvider({})
found, selector, record = await provider.check_dkim("example.com", ["google", "mail"])
found, selectors, record = await provider.check_dkim("example.com", ["google", "mail"])
assert found is False
assert selector is None
assert selectors == []
assert record is None
@@ -149,7 +166,7 @@ async def test_check_domain_all_present():
assert result.dmarc is True
assert result.spf is True
assert result.dkim is True
assert result.dkim_selector == "google"
assert result.dkim_selectors == ["google"]
@pytest.mark.asyncio
@@ -168,7 +185,7 @@ async def test_check_domain_uses_common_selectors_as_fallback():
provider = FakeDNSProvider({"default._domainkey.example.com": ["v=DKIM1; k=rsa; p=ABC"]})
result = await provider.check_domain("example.com", selectors=[])
assert result.dkim is True
assert result.dkim_selector == "default"
assert result.dkim_selectors == ["default"]
@pytest.mark.asyncio
@@ -178,7 +195,7 @@ async def test_check_domain_manual_selectors_take_priority():
provider = FakeDNSProvider({"custom._domainkey.example.com": ["v=DKIM1; k=rsa; p=XYZ"]})
result = await provider.check_domain("example.com", selectors=["custom"])
assert result.dkim is True
assert result.dkim_selector == "custom"
assert result.dkim_selectors == ["custom"]
# ---------------------------------------------------------------------------