facetted-interfaces/history/FIAM_v0.2_Draft6/tools/openapi_to_fiam.py

#!/usr/bin/env python3
"""OpenAPI 3.x -> FIAM v0.2-draft.4 declared-model ingest.

The importer is intentionally architecture-oriented, not a lossless OpenAPI converter.
It creates FIAM nodes/ports/contracts/operations/bindings/endpoints/resources/shapes,
plus evidence-backed relations. Detailed schema fields stay in the OpenAPI artifact.
"""
from __future__ import annotations
import argparse, hashlib, json, re
from pathlib import Path
from urllib.parse import urlparse
import yaml

HTTP_METHODS = ["get","put","post","delete","options","head","patch","trace"]
IGNORE_FINGERPRINT_KEYS = {
    "title","description","example","examples","externalDocs","deprecated","xml",
    "$comment","readOnly","writeOnly"
}

def safe(s: str) -> str:
    s = re.sub(r"[^A-Za-z0-9._:-]+", "-", str(s)).strip("-.")
    return s or "unnamed"

def load_doc(path: Path):
    text = path.read_text(encoding="utf-8")
    return json.loads(text) if path.suffix.lower()==".json" else yaml.safe_load(text)

def pointer_escape(s): return str(s).replace("~","~0").replace("/","~1")

def schema_pointer(name): return f"#/components/schemas/{pointer_escape(name)}"

def shape_kind(schema):
    if not isinstance(schema, dict): return "fiam.information-shape.opaque"
    if "$ref" in schema: return None
    if "oneOf" in schema or "anyOf" in schema: return "fiam.information-shape.union"
    typ=schema.get("type")
    if typ=="array": return "fiam.information-shape.collection"
    if typ=="object" or "properties" in schema or "additionalProperties" in schema: return "fiam.information-shape.object"
    if typ in {"string","number","integer","boolean","null"} or "enum" in schema: return "fiam.information-shape.scalar"
    return "fiam.information-shape.opaque"

def normalize_schema(x):
    if isinstance(x, dict):
        return {k:normalize_schema(v) for k,v in sorted(x.items()) if k not in IGNORE_FINGERPRINT_KEYS}
    if isinstance(x, list): return [normalize_schema(v) for v in x]
    return x

def fingerprint(schema):
    raw=json.dumps(normalize_schema(schema),sort_keys=True,separators=(",",":"),ensure_ascii=False)
    return hashlib.sha256(raw.encode()).hexdigest()

def media_rep(media):
    m=(media or "").lower().split(";",1)[0].strip()
    if m=="application/json" or m.endswith("+json"): return "fiam.representation.json"
    if m in {"application/xml","text/xml"} or m.endswith("+xml"): return "fiam.representation.xml"
    if m=="application/x-www-form-urlencoded": return "fiam.representation.form-urlencoded"
    if m.startswith("text/"): return "fiam.representation.text"
    if m in {"application/octet-stream","application/pdf","image/png","image/jpeg"}: return "fiam.representation.binary"
    return None

def deref_name(ref):
    p="#/components/schemas/"
    return ref[len(p):].replace("~1","/").replace("~0","~") if isinstance(ref,str) and ref.startswith(p) else None

class Importer:
    def __init__(self, doc, source_path, source_url=None):
        self.doc=doc; self.source_path=str(source_path)
        self.source_url=source_url or doc.get("x-source-url") or self.source_path
        title=doc.get("info",{}).get("title") or Path(source_path).stem
        self.slug=safe(title.lower())
        self.model={
            "fiam_version":"0.2",
            "instance":{
                "id":f"model.openapi.{self.slug}","kind":"model","title":f"OpenAPI ingest — {title}",
                "status":"provisional","intent":"Declared FIAM architecture model generated from an OpenAPI 3.x description."
            },
            "nodes":[],"ports":[],"contracts":[],"endpoints":[],"bindings":[],
            "information_resources":[],"information_shapes":[],"artifacts":[],"security":[],"assertions":[]
        }
        self.contract_id=f"contract.openapi.{self.slug}"
        self.node_id=f"node.openapi.{self.slug}"
        self.port_id=f"port.openapi.{self.slug}"
        self.artifact_id=f"artifact.openapi.{self.slug}"
        self.shape_ids={}; self.inline_shape_ids={}; self.op_ids=[]; self.binding_ids=[]
        self.assert_n=0

    def evidence(self, locator, mode="fiam.evidence.declared", confidence=1.0, source_type="openapi-document", metadata=None):
        md={"source_url":self.source_url}
        if metadata: md.update(metadata)
        return {"mode":mode,"source":{"type":source_type,"ref":self.artifact_id if source_type=="openapi-document" else source_type,
                                           "locator":locator,"metadata":md},
                "observer":"openapi-ingest/v0.2-draft.4","confidence":confidence,"stance":"supports"}

    def relation(self, subject,predicate,obj,locator,facets=None,mode="fiam.evidence.declared",confidence=1.0,source_type="openapi-document",metadata=None):
        self.assert_n+=1
        a={"id":f"assertion.openapi.{self.assert_n:04d}","assertion_kind":"relation","subject":subject,"predicate":predicate,
           "polarity":"asserted","object_ref":obj,"evidence":self.evidence(locator,mode,confidence,source_type,metadata)}
        if facets: a["facets"]=facets
        self.model["assertions"].append(a); return a

    def property(self, subject,predicate,value,locator,facets=None,mode="fiam.evidence.declared",confidence=1.0,source_type="openapi-document",metadata=None):
        self.assert_n+=1
        a={"id":f"assertion.openapi.{self.assert_n:04d}","assertion_kind":"property","subject":subject,"predicate":predicate,
           "polarity":"asserted","value":value,"evidence":self.evidence(locator,mode,confidence,source_type,metadata)}
        if facets: a["facets"]=facets
        self.model["assertions"].append(a); return a

    def component_shape(self,name,schema):
        if name in self.shape_ids: return self.shape_ids[name]
        sid=f"shape.openapi.{self.slug}.{safe(name)}"; self.shape_ids[name]=sid
        self.model["information_shapes"].append({"id":sid,"kind":"information-shape","name":name,
                                                  "shape_kind":shape_kind(schema) or "fiam.information-shape.opaque",
                                                  "facets":{"openapi_component":name}})
        return sid

    def ensure_inline_shape(self, schema, pointer, name_hint):
        if not isinstance(schema,dict): return None
        if "$ref" in schema:
            n=deref_name(schema["$ref"])
            return self.shape_ids.get(n) or (self.component_shape(n,self.doc.get("components",{}).get("schemas",{}).get(n,{})) if n else None)
        key=pointer
        if key in self.inline_shape_ids: return self.inline_shape_ids[key]
        sid=f"shape.openapi.{self.slug}.inline.{safe(name_hint)}.{hashlib.sha1(pointer.encode()).hexdigest()[:8]}"
        self.inline_shape_ids[key]=sid
        self.model["information_shapes"].append({"id":sid,"kind":"information-shape","name":name_hint,
                                                  "shape_kind":shape_kind(schema) or "fiam.information-shape.opaque",
                                                  "facets":{"openapi_pointer":pointer}})
        # Array element shape when meaningful.
        if schema.get("type")=="array" and isinstance(schema.get("items"),dict):
            child=self.ensure_inline_shape(schema["items"],pointer+"/items",name_hint+" element")
            if child:
                self.relation(sid,"fiam.relation.element-shape",child,pointer+"/items",
                              {"slot":"items","role":"fiam.shape-role.element"})
        return sid

    def build_component_shapes(self):
        schemas=self.doc.get("components",{}).get("schemas",{}) or {}
        # First pass: identity.
        for name,schema in schemas.items(): self.component_shape(name,schema or {})
        # Second pass: coarse composition, still no field lists in Core.
        for name,schema in schemas.items():
            sid=self.shape_ids[name]
            if not isinstance(schema,dict): continue
            if schema.get("type")=="array" and isinstance(schema.get("items"),dict):
                child=self.ensure_inline_shape(schema["items"],schema_pointer(name)+"/items",f"{name} element")
                if child: self.relation(sid,"fiam.relation.element-shape",child,schema_pointer(name)+"/items",{"slot":"items","role":"fiam.shape-role.element"})
            for prop,ps in (schema.get("properties") or {}).items():
                if not isinstance(ps,dict): continue
                pp=schema_pointer(name)+"/properties/"+pointer_escape(prop)
                target=None; slot=prop
                if "$ref" in ps:
                    target=self.ensure_inline_shape(ps,pp,prop)
                elif ps.get("type")=="array" and isinstance(ps.get("items"),dict):
                    # Preserve array-ness as its own shape, then link element type.
                    target=self.ensure_inline_shape(ps,pp,f"{name}.{prop}")
                elif ps.get("type") in {"object","array"} or "oneOf" in ps or "anyOf" in ps:
                    target=self.ensure_inline_shape(ps,pp,f"{name}.{prop}")
                if target:
                    self.relation(sid,"fiam.relation.embeds-shape",target,pp,{"slot":slot,"role":"fiam.shape-role.envelope-member"})
        # Source artifact describes all component and inline shapes plus contract.

    def identity_candidates(self):
        schemas=self.doc.get("components",{}).get("schemas",{}) or {}
        groups={}
        for name,schema in schemas.items(): groups.setdefault(fingerprint(schema or {}),[]).append(name)
        for fp,names in groups.items():
            if len(names)<2: continue
            for i in range(len(names)):
                for j in range(i+1,len(names)):
                    a,b=names[i],names[j]
                    # structural equality is a candidate alignment only; never same-entity-as.
                    self.relation(self.shape_ids[a],"fiam.relation.corresponds-to",self.shape_ids[b],
                                  f"fingerprint:sha256:{fp}",mode="fiam.evidence.inferred",confidence=0.75,
                                  source_type="schema-structural-fingerprint",
                                  metadata={"left":schema_pointer(a),"right":schema_pointer(b),"algorithm":"normalized-openapi-schema-sha256"})


    def security_mapping(self, scheme):
        typ=str((scheme or {}).get("type") or "")
        if typ=="apiKey": return "fiam.security.authentication.api-key", ["fiam.credential.api-key"], None
        if typ=="http":
            hs=str(scheme.get("scheme") or "").lower()
            if hs=="basic": return "fiam.security.authentication.http-basic", ["fiam.credential.basic-credentials"], None
            if hs=="bearer": return "fiam.security.authentication.bearer-token", ["fiam.credential.bearer-token"], None
            return "fiam.security.authentication.unspecified", [], None
        if typ=="oauth2": return "fiam.security.authentication.oauth2", ["fiam.credential.access-token"], None
        if typ=="openIdConnect": return "fiam.security.authentication.openid-connect", ["fiam.credential.id-token","fiam.credential.access-token"], None
        if typ=="mutualTLS": return "fiam.security.authentication.mutual-tls", ["fiam.credential.client-certificate"], "fiam.security.tls"
        return "fiam.security.authentication.unspecified", [], None

    def apply_security_requirements(self, applies_to, requirements, pointer, scope_kind):
        """Map OpenAPI Security Requirement alternatives without flattening OR/AND semantics.

        The array is OR. Multiple scheme names inside one requirement object are AND.
        One FIAM SecurityProfile is emitted per referenced scheme with a shared
        requirement_alternative facet.
        """
        if requirements is None: return
        if requirements==[]:
            self.property(applies_to,"fiam.property.security-override","none",pointer)
            return
        schemes=(self.doc.get("components",{}).get("securitySchemes",{}) or {})
        anonymous_alt=any(isinstance(r,dict) and not r for r in requirements)
        state="anonymous-alternative" if anonymous_alt else ("operation" if scope_kind=="operation" else "declared")
        self.property(applies_to,"fiam.property.security-override",state,pointer)
        for alt, req in enumerate(requirements, start=1):
            if not isinstance(req,dict) or not req: continue
            for name, scopes in req.items():
                sd=schemes.get(name)
                if not isinstance(sd,dict) or "$ref" in sd: continue
                auth, creds, transport=self.security_mapping(sd)
                sid=f"security.openapi.{self.slug}.{safe(applies_to)}.{alt}.{safe(name)}"
                facets={"openapi_security_scheme":name,"openapi_security_type":sd.get("type"),"requirement_alternative":alt,"requirement_operator":"and-within-alternative","required_scopes":list(scopes or [])}
                for k in ("name","in","scheme","bearerFormat","openIdConnectUrl"):
                    if sd.get(k) is not None: facets["openapi_"+k]=sd.get(k)
                if sd.get("flows"): facets["openapi_oauth_flows"]=sorted(sd["flows"].keys())
                prof={"id":sid,"applies_to":applies_to,"authentication":auth,"facets":facets}
                if creds: prof["credential_artifacts"]=creds
                if transport: prof["transport_security"]=transport
                if not any(x.get("id")==sid for x in self.model["security"]): self.model["security"].append(prof)
                # Evidence that the security profile is declared by the source description.
                self.property(sid,"fiam.property.security-override","required",pointer+"/"+str(alt-1)+"/"+pointer_escape(name),facets={"applies_to":applies_to})

    def add_shape_participation(self, opid, binding_ids, pred, schema, pointer, media, role):
        sid=self.ensure_inline_shape(schema,pointer,opid.split('.')[-1]+" "+role.split('.')[-1])
        if not sid: return
        self.relation(opid,pred,sid,pointer,{"role":role})
        rep=media_rep(media)
        for bid in binding_ids:
            facets={"operation_ref":opid,"role":role}
            if rep: facets["representation"]=rep
            self.relation(bid,pred,sid,pointer,facets)

    def build(self):
        info=self.doc.get("info",{})
        self.model["nodes"].append({"id":self.node_id,"kind":"system-node","name":info.get("title",self.slug),"node_type":"fiam.node-type.service"})
        self.model["ports"].append({"id":self.port_id,"owner":self.node_id,"participations":[{"contract":self.contract_id,"mode":"offers"}]})
        contract={"id":self.contract_id,"kind":"interface-contract","name":info.get("title",self.slug),"contract_forms":["fiam.contract-form.openapi"],"operations":[]}
        self.model["contracts"].append(contract)
        self.build_component_shapes()
        self.apply_security_requirements(self.contract_id,self.doc.get("security"),"#/security","contract")

        # Endpoints/bindings from servers. If none, keep an unlocated HTTP binding.
        servers=self.doc.get("servers") or [{}]
        for idx,server in enumerate(servers):
            url=server.get("url") if isinstance(server,dict) else None
            eid=f"endpoint.openapi.{self.slug}.{idx+1}"; bid=f"binding.openapi.{self.slug}.{idx+1}"
            ep={"id":eid,"kind":"endpoint","name":server.get("description",f"OpenAPI server {idx+1}") if isinstance(server,dict) else f"OpenAPI server {idx+1}"}
            if url:
                ep["address"]=url
                if urlparse(url).scheme.lower()=="https": ep["address_scheme"]="fiam.address-scheme.https-url"
            self.model["endpoints"].append(ep)
            stack=[{"role":"fiam.stack-role.application-transfer-protocol","term":"fiam.protocol.http"}]
            if url and urlparse(url).scheme.lower()=="https": stack += [{"role":"fiam.stack-role.security-protocol","term":"fiam.protocol.tls"}]
            b={"id":bid,"kind":"binding","contract":self.contract_id,"provider_port":self.port_id,"endpoints":[eid],"communication_stack":stack,"operations":[]}
            self.model["bindings"].append(b); self.binding_ids.append(bid)
            self.property(bid,"fiam.property.communication-protocol","fiam.protocol.http",f"#/servers/{idx}")
            if url:
                pu=urlparse(url)
                if pu.hostname:
                    self.property(eid,"fiam.property.server-address",pu.hostname,f"#/servers/{idx}/url")
                if pu.port:
                    self.property(eid,"fiam.property.server-port",pu.port,f"#/servers/{idx}/url")

        for path,pitem in (self.doc.get("paths") or {}).items():
            if not isinstance(pitem,dict): continue
            # Path template is a logical information resource selected through the HTTP API.
            collection = not bool(re.search(r"\{[^}]+\}\s*$",path))
            rid=f"resource.openapi.{self.slug}.{safe(path)}"
            self.model["information_resources"].append({
                "id":rid,"kind":"information-resource","owner":self.node_id,"name":path,
                "resource_kind":"fiam.information-resource.api-resource-collection" if collection else "fiam.information-resource.api-resource-object",
                "resource_scope":"fiam.resource-scope.logical-selection",
                "selectors":[{"kind":"fiam.resource-selector.http-path-template","value":path}]
            })
            for method in HTTP_METHODS:
                op=pitem.get(method)
                if not isinstance(op,dict): continue
                opkey=op.get("operationId") or f"{method}-{path}"
                opid=f"operation.openapi.{self.slug}.{safe(opkey)}"; self.op_ids.append(opid)
                effect="fiam.effect.read" if method in {"get","head","options"} else "fiam.effect.write"
                purpose=["fiam.purpose.query"] if effect=="fiam.effect.read" else ["fiam.purpose.mutation"]
                o={"id":opid,"kind":"operation","name":op.get("summary") or opkey,"description":op.get("description",""),
                   "interaction_style":"fiam.interaction-style.request-response","effect":effect,"purpose":purpose,
                   "x-http-method":method.upper(),"x-openapi-path":path,"x-openapi-operation-id":op.get("operationId")}
                if o["x-openapi-operation-id"] is None: o.pop("x-openapi-operation-id")
                contract["operations"].append(o)
                for b in self.model["bindings"]: b["operations"].append(opid)
                op_ptr=f"#/paths/{pointer_escape(path)}/{method}"
                if "security" in op: self.apply_security_requirements(opid,op.get("security"),op_ptr+"/security","operation")
                self.property(opid,"fiam.property.http-method",method.upper(),op_ptr)
                self.property(opid,"fiam.property.http-route",path,op_ptr)
                self.relation(opid,"fiam.relation.targets",rid,op_ptr)
                # Request body
                rb=op.get("requestBody") or {}
                if isinstance(rb,dict) and "$ref" not in rb:
                    for media,md in (rb.get("content") or {}).items():
                        if isinstance(md,dict) and isinstance(md.get("schema"),dict):
                            self.add_shape_participation(opid,self.binding_ids,"fiam.relation.accepts-shape",md["schema"],op_ptr+"/requestBody/content/"+pointer_escape(media)+"/schema",media,"fiam.shape-role.request")
                # Responses
                for status,resp in (op.get("responses") or {}).items():
                    if not isinstance(resp,dict) or "$ref" in resp: continue
                    pred="fiam.relation.produces-shape" if str(status).startswith("2") else "fiam.relation.errors-with-shape"
                    role="fiam.shape-role.response" if pred.endswith("produces-shape") else "fiam.shape-role.error"
                    for media,md in (resp.get("content") or {}).items():
                        if isinstance(md,dict) and isinstance(md.get("schema"),dict):
                            self.add_shape_participation(opid,self.binding_ids,pred,md["schema"],op_ptr+"/responses/"+pointer_escape(status)+"/content/"+pointer_escape(media)+"/schema",media,role)

        self.identity_candidates()
        described=[self.contract_id]+[x["id"] for x in self.model["information_shapes"]]+[x["id"] for x in self.model.get("security",[])]
        self.model["artifacts"].append({"id":self.artifact_id,"kind":"artifact","name":Path(self.source_path).name,
                                         "artifact_type":"fiam.artifact.api-description","describes":described,"locator":self.source_url,
                                         "facets":{"openapi_version":str(self.doc.get("openapi","")),"document_version":str(info.get("version","")),"source_file":Path(self.source_path).name,
                                                   **({"source_note":self.doc.get("x-source-note")} if self.doc.get("x-source-note") else {})}})
        # strip empty sections
        self.model={k:v for k,v in self.model.items() if not isinstance(v,list) or v}
        return self.model

def main():
    ap=argparse.ArgumentParser(description=__doc__)
    ap.add_argument("input",type=Path); ap.add_argument("-o","--output",type=Path,required=True); ap.add_argument("--source-url")
    args=ap.parse_args(); doc=load_doc(args.input)
    if not isinstance(doc,dict) or not str(doc.get("openapi","")).startswith("3."):
        raise SystemExit("Only OpenAPI 3.x documents are supported in draft.4")
    out=Importer(doc,args.input,args.source_url).build(); args.output.parent.mkdir(parents=True,exist_ok=True)
    args.output.write_text(yaml.safe_dump(out,sort_keys=False,allow_unicode=True,width=120),encoding="utf-8")
    print(f"wrote {args.output}: {len(out.get('contracts',[{}])[0].get('operations',[]))} operations, {len(out.get('information_shapes',[]))} shapes, {len(out.get('assertions',[]))} assertions")
if __name__=="__main__": main()