Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 21 additions & 14 deletions docs/AUTHORING.md
Original file line number Diff line number Diff line change
Expand Up @@ -204,13 +204,15 @@ To support a launched provider, pass its generated `ClientPoint` when constructi
host.WithClientProviders(mypointpb.ClientPoint),
```

`ClientPoint` builds an in-process caller from the gRPC connection.
`ClientPoint` builds a typed in-process caller from the gRPC connection, so daemon code calls a launched provider like a built-in one.
Wire it only for Points the daemon itself calls.
This list is the wiring boundary for internal calls to launched providers: an unlisted declared point is rejected unless it is offered only for external publication.
Use `host.WithProviderPolicy` to restrict which extensions may provide internally wired points.
See [DESIGN.md#discovery-security](./DESIGN.md#discovery-security).

An offered-only process Point does not need `ClientPoint` wiring because the Host
proxies its generated service without constructing an internal caller.
A Point that only external callers use does not need `ClientPoint` wiring.
Offer it for publication instead: the proxy forwards calls to the extension by gRPC service name without decoding them.
For example, a naming Point that the daemon calls needs `ClientPoint`, while a job API that only external tools call needs only an offer.

### Publishing an ordinary Point

Expand Down Expand Up @@ -250,17 +252,22 @@ _, err := host.New(ctx, host.WithProviderPolicy(host.PointPolicyFunc(func(identi
})))
```

Publication authorization uses `servicev0.Point.ID()` and receives the
host-attested identity. Internal provider admission uses the same policy with
ordinary Point IDs and different nil behavior: a nil policy preserves all
internally wired providers while dropping publication, while a non-nil policy
can allow, drop, or reject each use by identity and Point. `Drop` omits an
ordinary provider without unloading its extension; `Reject` fails Host
construction and preserves its error cause. A zero result and a typed nil
`PointPolicyFunc` reject the request. Offered-only process Points without
`ClientPoint` wiring consult the policy only with `servicev0.Point.ID()`.
The `servicev0` offer marker itself is exempt from provider admission and
governs publication of the complete validated offered-point set.
Publication authorization uses `servicev0.Point.ID()` and receives the host-attested identity.
Internal provider admission uses the same policy with ordinary Point IDs.
A nil policy allows every internally wired provider and drops all publication.
A non-nil policy decides each use by identity and Point:

- `Allow` admits the provider or publication.
- `Drop` omits an ordinary provider and its offer.
- `Reject` fails Host construction and preserves its error cause.

A zero result and a typed nil `PointPolicyFunc` reject the request.
The `servicev0` offer marker itself is exempt from provider admission.
Offered-only process Points have no internal provider, so the policy sees them only through `servicev0.Point.ID()`.

If policy drops every ordinary provider of an extension, the Host skips the extension instead of initializing it.
A launched extension stays loaded while it publishes an offered-only Point, because external callers still reach that Point through the proxy.
For example, when policy drops the naming Point an extension provides to the daemon but allows publication, the extension keeps serving its offered-only job API.

For an in-process extension, supply generated adapters at Host composition:

Expand Down
219 changes: 160 additions & 59 deletions host/host.go
Original file line number Diff line number Diff line change
Expand Up @@ -55,8 +55,8 @@ func Allow() PointPolicyResult {
return PointPolicyResult{action: pointPolicyActionAllow}
}

// Drop returns a decision that silently omits an ordinary provider or leaves an
// offered set private.
// Drop omits an ordinary provider (including its offer), or keeps all offers
// private when applied to servicev0.Point.ID().
func Drop() PointPolicyResult {
return PointPolicyResult{action: pointPolicyActionDrop}
}
Expand Down Expand Up @@ -85,8 +85,8 @@ func (result PointPolicyResult) resolve() (pointPolicyAction, error) {
}

// PointPolicy decides whether an extension identity may provide a Point.
// A decision on servicev0.Point.ID() controls publication of the complete
// offered set.
// A decision on servicev0.Point.ID() controls publication of offered points
// whose providers were not dropped by policy.
// A nil policy allows ordinary providers and drops publication.
type PointPolicy interface {
Decide(identity extensions.ExtensionIdentity, point extensions.PointID) PointPolicyResult
Expand Down Expand Up @@ -237,6 +237,14 @@ type registeredExtension struct {
providers []extensions.Provider
}

type publicationState struct {
policy PointPolicy
servers map[extensions.PointID]serverpoint.Registration
published map[extensions.ExtensionID]map[extensions.PointID][]string
owners map[string]extensions.ExtensionID
reserved map[string]bool
}

// hostedExtension is the runtime-neutral declaration and lifecycle surface
// needed to adapt an externally hosted extension to the broker.
type hostedExtension struct {
Expand Down Expand Up @@ -307,8 +315,13 @@ func New(ctx context.Context, optionList ...Option) (_ *Host, retErr error) {
conns := make(map[extensions.ExtensionID]grpc.ClientConnInterface)
var loaded []loadedExtension
var registered []registeredExtension
publishedServices := make(map[extensions.ExtensionID]map[extensions.PointID][]string)
publishedOwners := make(map[string]extensions.ExtensionID)
publication := publicationState{
policy: policy,
servers: pointServers,
published: make(map[extensions.ExtensionID]map[extensions.PointID][]string),
owners: make(map[string]extensions.ExtensionID),
reserved: reservedServices,
}
var inProcessServices []servicegrpc.Service
var callback *grpc.Server
// The broker only shuts down initialized extensions, so construction failures
Expand Down Expand Up @@ -343,7 +356,23 @@ func New(ctx context.Context, optionList ...Option) (_ *Host, retErr error) {
if err != nil {
return nil, err
}
services, err := collectInProcessPublications(identity, ext, policy, pointServers, publishedServices, publishedOwners, reservedServices)
if err := broker.ValidateDeclaration(decl); err != nil {
return nil, err
}
if extensionFullyDropped(decl.Providers, admittedProviders) {
// Dropping the extension must not hide malformed offers or a policy rejection.
offered, err := offeredInProcessPoints(decl)
if err != nil {
return nil, err
}
if len(offered) > 0 {
if _, err := publicationPolicyAction(identity, policy); err != nil {
return nil, err
}
}
continue
}
services, err := collectInProcessPublications(identity, ext, &publication, droppedProviderPoints(decl.Providers, admittedProviders))
if err != nil {
return nil, err
}
Expand Down Expand Up @@ -379,9 +408,22 @@ func New(ctx context.Context, optionList ...Option) (_ *Host, retErr error) {
if err != nil {
return nil, err
}
if err := approveProcessPublications(identity, started, policy, publishedServices, publishedOwners, reservedServices); err != nil {
if err := broker.ValidateDeclaration(decl); err != nil {
return nil, err
}
if err := approveProcessPublications(identity, started, &publication, droppedProviderPoints(decl.Providers, admittedProviders)); err != nil {
return nil, err
}
// Offered-only points have no provider for policy to drop, so they keep
// the process only when their offer is published.
if extensionFullyDropped(decl.Providers, admittedProviders) && len(publication.published[identity.ID]) == 0 {
// The process was launched for Describe; stop it before returning the host.
if err := started.Close(context.WithoutCancel(ctx)); err != nil {
return nil, fmt.Errorf("close dropped extension %q: %w", identity.ID, err)
}
loaded = loaded[:len(loaded)-1]
continue
}
decl.Providers = admittedProviders
if err := b.Register(identity, extensions.New(decl)); err != nil {
return nil, err
Expand Down Expand Up @@ -433,7 +475,7 @@ func New(ctx context.Context, optionList ...Option) (_ *Host, retErr error) {
}
log.G(ctx).WithFields(fields).Info("loaded extension")
}
return &Host{broker: b, conns: conns, loaded: loaded, publishedServices: publishedServices, inProcessServices: inProcessServices, callback: callback}, nil
return &Host{broker: b, conns: conns, loaded: loaded, publishedServices: publication.published, inProcessServices: inProcessServices, callback: callback}, nil
}

func validateHostIdentity(identity extensions.ExtensionIdentity, decl extensions.Declaration) error {
Expand Down Expand Up @@ -470,45 +512,131 @@ func admitProviders(identity extensions.ExtensionIdentity, providers []extension
return admitted, nil
}

func approveProcessPublications(identity extensions.ExtensionIdentity, started *launcher.Launched, policy PointPolicy, published map[extensions.ExtensionID]map[extensions.PointID][]string, owners map[string]extensions.ExtensionID, reserved map[string]bool) error {
// extensionFullyDropped ignores metadata providers and keeps extensions that
// declared no ordinary providers: policy had nothing to drop from them.
func extensionFullyDropped(declared, admitted []extensions.Provider) bool {
hadProvider := false
for _, provider := range declared {
if !extensions.IsMetadataPoint(provider.Point) {
hadProvider = true
break
}
}
if !hadProvider {
return false
}
for _, provider := range admitted {
if !extensions.IsMetadataPoint(provider.Point) {
return false
}
}
return true
}

// droppedProviderPoints excludes offered-only process points, which have no
// internal provider for policy to drop and remain governed by servicev0.
func droppedProviderPoints(declared, admitted []extensions.Provider) map[extensions.PointID]bool {
if len(declared) == len(admitted) {
return nil
}
dropped := make(map[extensions.PointID]bool, len(declared)-len(admitted))
for _, provider := range declared {
dropped[provider.Point] = true
}
for _, provider := range admitted {
delete(dropped, provider.Point)
}
return dropped
}

func approveProcessPublications(identity extensions.ExtensionIdentity, started *launcher.Launched, publication *publicationState, dropped map[extensions.PointID]bool) error {
if len(started.OfferedPoints) == 0 {
return nil
}
result := Drop()
if policy != nil {
result = policy.Decide(identity, servicev0.Point.ID())
action, err := publicationPolicyAction(identity, publication.policy)
if err != nil {
return err
}
action, cause := result.resolve()
switch action {
case pointPolicyActionDrop:
if action == pointPolicyActionDrop {
return nil
case pointPolicyActionReject:
return policyRejectionError("publish offered points", identity, servicev0.Point.ID(), cause)
}
for _, point := range started.OfferedPoints {
if dropped[point] {
continue
}
names := started.ProviderServices[point]
if len(names) == 0 {
return fmt.Errorf("extension %q offered point %q without a gRPC service", identity.ID, point)
}
for _, service := range names {
if reserved[service] {
if publication.reserved[service] {
return fmt.Errorf("extension %q cannot publish reserved gRPC service %q", identity.ID, service)
}
if owner, exists := owners[service]; exists {
if owner, exists := publication.owners[service]; exists {
return fmt.Errorf("extensions %q and %q both publish gRPC service %q", owner, identity.ID, service)
}
owners[service] = identity.ID
publication.owners[service] = identity.ID
}
if published[identity.ID] == nil {
published[identity.ID] = make(map[extensions.PointID][]string)
if publication.published[identity.ID] == nil {
publication.published[identity.ID] = make(map[extensions.PointID][]string)
}
published[identity.ID][point] = append([]string(nil), names...)
publication.published[identity.ID][point] = append([]string(nil), names...)
}
return nil
}

func collectInProcessPublications(identity extensions.ExtensionIdentity, ext extensions.Extension, policy PointPolicy, servers map[extensions.PointID]serverpoint.Registration, published map[extensions.ExtensionID]map[extensions.PointID][]string, owners map[string]extensions.ExtensionID, reserved map[string]bool) ([]servicegrpc.Service, error) {
func collectInProcessPublications(identity extensions.ExtensionIdentity, ext extensions.Extension, publication *publicationState, dropped map[extensions.PointID]bool) ([]servicegrpc.Service, error) {
decl := ext.Declaration()
offered, err := offeredInProcessPoints(decl)
if err != nil {
return nil, err
}
if len(offered) == 0 {
return nil, nil
}
action, err := publicationPolicyAction(identity, publication.policy)
if err != nil {
return nil, err
}
if action == pointPolicyActionDrop {
return nil, nil
}

providers := make(map[extensions.PointID]any, len(decl.Providers))
for _, provider := range decl.Providers {
providers[provider.Point] = provider.Impl
}
var services []servicegrpc.Service
for _, point := range offered {
if dropped[point] {
continue
}
impl := providers[point]
registration, ok := publication.servers[point]
if !ok {
return nil, fmt.Errorf("extension %q: allowed in-process offer for point %q has no server registration", decl.ID, point)
}
service, err := servicegrpc.Adapt(registration, impl)
if err != nil {
return nil, fmt.Errorf("extension %q: publish point %q: %w", decl.ID, point, err)
}
if publication.reserved[service.Name] {
return nil, fmt.Errorf("extension %q cannot publish reserved gRPC service %q", decl.ID, service.Name)
}
if owner, exists := publication.owners[service.Name]; exists {
return nil, fmt.Errorf("extensions %q and %q both publish gRPC service %q", owner, decl.ID, service.Name)
}
publication.owners[service.Name] = decl.ID
services = append(services, service)
if publication.published[decl.ID] == nil {
publication.published[decl.ID] = make(map[extensions.PointID][]string)
}
publication.published[decl.ID][point] = []string{service.Name}
}
return services, nil
}

func offeredInProcessPoints(decl extensions.Declaration) ([]extensions.PointID, error) {
providers := make(map[extensions.PointID]any, len(decl.Providers))
for _, provider := range decl.Providers {
if _, exists := providers[provider.Point]; exists {
Expand Down Expand Up @@ -545,46 +673,19 @@ func collectInProcessPublications(identity extensions.ExtensionIdentity, ext ext
offered = append(offered, point)
}
}
if len(offered) == 0 {
return nil, nil
}
return offered, nil
}

func publicationPolicyAction(identity extensions.ExtensionIdentity, policy PointPolicy) (pointPolicyAction, error) {
result := Drop()
if policy != nil {
result = policy.Decide(identity, servicev0.Point.ID())
}
action, cause := result.resolve()
switch action {
case pointPolicyActionDrop:
return nil, nil
case pointPolicyActionReject:
return nil, policyRejectionError("publish offered points", identity, servicev0.Point.ID(), cause)
if action == pointPolicyActionReject {
return action, policyRejectionError("publish offered points", identity, servicev0.Point.ID(), cause)
}

var services []servicegrpc.Service
for _, point := range offered {
impl := providers[point]
registration, ok := servers[point]
if !ok {
return nil, fmt.Errorf("extension %q: allowed in-process offer for point %q has no server registration", decl.ID, point)
}
service, err := servicegrpc.Adapt(registration, impl)
if err != nil {
return nil, fmt.Errorf("extension %q: publish point %q: %w", decl.ID, point, err)
}
if reserved[service.Name] {
return nil, fmt.Errorf("extension %q cannot publish reserved gRPC service %q", decl.ID, service.Name)
}
if owner, exists := owners[service.Name]; exists {
return nil, fmt.Errorf("extensions %q and %q both publish gRPC service %q", owner, decl.ID, service.Name)
}
owners[service.Name] = decl.ID
services = append(services, service)
if published[decl.ID] == nil {
published[decl.ID] = make(map[extensions.PointID][]string)
}
published[decl.ID][point] = []string{service.Name}
}
return services, nil
return action, nil
}

func policyRejectionError(operation string, identity extensions.ExtensionIdentity, point extensions.PointID, cause error) error {
Expand Down
Loading
Loading