"""DB-backed tests for live certificate owner resolution. Ownership follows the DNS record / zone domain the cert serves — the cert no longer holds a frozen owner snapshot. These tests pin the resolution chain: manual override → DNS record owner → domain owner → network owner fallback, plus the dormant-legacy-snapshot behaviour. """ from __future__ import annotations import uuid from datetime import UTC, datetime, timedelta import pytest import pytest_asyncio from sqlalchemy.ext.asyncio import AsyncSession from app.domains.cmdb.cert_owner_resolver import ( OWNER_SOURCE_DNS_RECORD, OWNER_SOURCE_DOMAIN, OWNER_SOURCE_MANUAL, OWNER_SOURCE_NETWORK, OWNER_SOURCE_NONE, CertOwnerResolver, ) from app.domains.cmdb.models import ( CERT_SOURCE_MANUAL, CERT_SOURCE_PROBED, CmdbCertificate, CmdbDnsRecord, CmdbDomain, ) from app.identity.person_models import IdentityPerson from app.integrations.models import Integration async def _person(db: AsyncSession, name: str) -> IdentityPerson: p = IdentityPerson( id=uuid.uuid4(), source_system="successfactors", hris_person_id=f"sf-{uuid.uuid4().hex[:8]}", display_name=name, work_email=f"{name.lower()}-{uuid.uuid4().hex[:6]}@pingcap.com", employment_status="active", ) db.add(p) # Flush so the row exists before any string-column FK references it # (no ORM relationship links these, so the unit-of-work can't infer # the insert ordering on its own). await db.flush() return p @pytest_asyncio.fixture async def integ(db_session: AsyncSession) -> Integration: i = Integration( id=uuid.uuid4(), kind="cloud", provider="aliyun", name="pytest-resolver", slug=f"pytest-resolver-{uuid.uuid4().hex[:8]}", enabled=True, is_primary=False, capabilities=["domains", "dns_records"], ) db_session.add(i) await db_session.commit() return i def _probed(host: str, **kw) -> CmdbCertificate: return CmdbCertificate( id=uuid.uuid4(), dedup_key=f"probed:{host}:443", source=CERT_SOURCE_PROBED, probe_host=host, probe_port=443, common_name=kw.pop("common_name", host), sans=kw.pop("sans", [host]), not_after=datetime.now(UTC) + timedelta(days=30), status="valid", first_seen_at=datetime.now(UTC), **kw, ) @pytest.mark.asyncio async def test_record_owner_is_primary_domain_owner_is_escalation( db_session: AsyncSession, integ: Integration, ): rec_owner = await _person(db_session, "RecOwner") dom_owner = await _person(db_session, "DomOwner") db_session.add(CmdbDomain( id=uuid.uuid4(), integration_id=integ.id, domain_name="example.com", owner_person_id=str(dom_owner.id), )) db_session.add(CmdbDnsRecord( integration_id=integ.id, zone_name="example.com", record_id="rr1", name="www", record_type="A", value="1.2.3.4", status="enable", owner_person_id=str(rec_owner.id), )) await db_session.commit() resolver = await CertOwnerResolver.build(db_session) res = resolver.resolve(_probed("www.example.com")) # 本人 = DNS record owner; 域名负责人 = domain owner (escalation target). assert res.owner_person_id == str(rec_owner.id) assert res.owner_source == OWNER_SOURCE_DNS_RECORD assert res.domain_owner_person_id == str(dom_owner.id) assert res.matched_zone == "example.com" assert res.owner_is_manual is False @pytest.mark.asyncio async def test_falls_back_to_domain_owner_when_record_unowned( db_session: AsyncSession, integ: Integration, ): dom_owner = await _person(db_session, "DomOwner") db_session.add(CmdbDomain( id=uuid.uuid4(), integration_id=integ.id, domain_name="example.com", owner_person_id=str(dom_owner.id), )) db_session.add(CmdbDnsRecord( integration_id=integ.id, zone_name="example.com", record_id="rr2", name="api", record_type="CNAME", value="lb.example.com", status="enable", owner_person_id=None, )) await db_session.commit() resolver = await CertOwnerResolver.build(db_session) res = resolver.resolve(_probed("api.example.com")) assert res.owner_person_id == str(dom_owner.id) assert res.owner_source == OWNER_SOURCE_DOMAIN assert res.domain_owner_person_id == str(dom_owner.id) @pytest.mark.asyncio async def test_network_owner_fallback_then_none( db_session: AsyncSession, integ: Integration, ): net_owner = await _person(db_session, "NetOwner") # A domain with NO owner and no matching record. db_session.add(CmdbDomain( id=uuid.uuid4(), integration_id=integ.id, domain_name="orphan.com", owner_person_id=None, )) await db_session.commit() # With a network owner configured → 兜底. resolver = await CertOwnerResolver.build( db_session, network_owner_id=str(net_owner.id), ) res = resolver.resolve(_probed("svc.orphan.com")) assert res.owner_person_id == str(net_owner.id) assert res.owner_source == OWNER_SOURCE_NETWORK assert res.domain_owner_person_id is None # Without a network owner → truly ownerless. resolver2 = await CertOwnerResolver.build(db_session) res2 = resolver2.resolve(_probed("svc.orphan.com")) assert res2.owner_person_id is None assert res2.owner_source == OWNER_SOURCE_NONE @pytest.mark.asyncio async def test_manual_override_wins_legacy_snapshot_ignored( db_session: AsyncSession, integ: Integration, ): rec_owner = await _person(db_session, "RecOwner") override = await _person(db_session, "Override") legacy = await _person(db_session, "LegacySnapshot") db_session.add(CmdbDomain( id=uuid.uuid4(), integration_id=integ.id, domain_name="example.com", owner_person_id=str(rec_owner.id), )) db_session.add(CmdbDnsRecord( integration_id=integ.id, zone_name="example.com", record_id="rr3", name="app", record_type="A", value="1.2.3.4", status="enable", owner_person_id=str(rec_owner.id), )) await db_session.commit() resolver = await CertOwnerResolver.build(db_session) # Manual override (owner_is_manual=True) beats the DNS chain. manual_cert = _probed( "app.example.com", owner_person_id=str(override.id), owner_is_manual=True, ) res = resolver.resolve(manual_cert) assert res.owner_person_id == str(override.id) assert res.owner_source == OWNER_SOURCE_MANUAL assert res.owner_is_manual is True # Legacy snapshot (owner set but owner_is_manual=False) is IGNORED — # the resolver live-resolves to the DNS record owner instead. legacy_cert = _probed( "app.example.com", owner_person_id=str(legacy.id), owner_is_manual=False, ) res2 = resolver.resolve(legacy_cert) assert res2.owner_person_id == str(rec_owner.id) assert res2.owner_source == OWNER_SOURCE_DNS_RECORD @pytest.mark.asyncio async def test_cloud_cert_without_probe_host_attributes_by_names( db_session: AsyncSession, integ: Integration, ): dom_owner = await _person(db_session, "DomOwner") db_session.add(CmdbDomain( id=uuid.uuid4(), integration_id=integ.id, domain_name="pingcap.net", owner_person_id=str(dom_owner.id), )) await db_session.commit() resolver = await CertOwnerResolver.build(db_session) # A cloud/manual wildcard cert (no probe_host) attributes via CN/SANs. cloud_cert = CmdbCertificate( id=uuid.uuid4(), dedup_key=f"{CERT_SOURCE_MANUAL}:{uuid.uuid4()}", source=CERT_SOURCE_MANUAL, common_name="*.pingcap.net", sans=["*.pingcap.net"], probe_host=None, not_after=datetime.now(UTC) + timedelta(days=10), status="valid", first_seen_at=datetime.now(UTC), ) res = resolver.resolve(cloud_cert) assert res.matched_zone == "pingcap.net" assert res.owner_person_id == str(dom_owner.id) assert res.owner_source == OWNER_SOURCE_DOMAIN @pytest.mark.asyncio async def test_longest_zone_suffix_wins( db_session: AsyncSession, integ: Integration, ): apex_owner = await _person(db_session, "ApexOwner") sub_owner = await _person(db_session, "SubOwner") db_session.add_all([ CmdbDomain( id=uuid.uuid4(), integration_id=integ.id, domain_name="pingcap.net", owner_person_id=str(apex_owner.id), ), CmdbDomain( id=uuid.uuid4(), integration_id=integ.id, domain_name="wiki.pingcap.net", owner_person_id=str(sub_owner.id), ), ]) await db_session.commit() resolver = await CertOwnerResolver.build(db_session) # tidb.wiki.pingcap.net matches the MORE SPECIFIC wiki.pingcap.net zone. res = resolver.resolve(_probed("tidb.wiki.pingcap.net")) assert res.matched_zone == "wiki.pingcap.net" assert res.owner_person_id == str(sub_owner.id)