package asterisk import ( "context" "errors" "fmt" "net" "net/netip" "strconv" "strings" "sync" "sync/atomic" "time" "git.ipao.vip/rogee/go-sip/internal/media" "github.com/CyCoreSystems/ari/v5" "github.com/CyCoreSystems/ari/v5/client/native" ) // NativeDial contains only the Dispatcher-selected SIP identity and the // Agent's explicitly configured SLIN16 RTP payload profile. type NativeDial struct { ExecutionID, TrunkID, DialedCallee, CallerID string AnswerTimeout time.Duration MediaPayloadType uint8 BindSIPCall func(executionID, callID string) error // before Dial; no number/time matching } // NativeCallFailure records a secret-free stage for a possibly ambiguous ARI // attempt. The wrapped cause remains available for programmatic inspection. type NativeCallFailure struct { Phase string Cause error ConfirmedEnd bool // only after a dialed pre-answer channel is verified absent from Asterisk } func (e *NativeCallFailure) Error() string { if errors.Is(e.Cause, context.DeadlineExceeded) { return fmt.Sprintf("native ARI %s deadline: %T", e.Phase, e.Cause) } return fmt.Sprintf("native ARI %s outcome unknown: %T", e.Phase, e.Cause) } func (e *NativeCallFailure) Unwrap() error { return e.Cause } func NativeCallPhase(err error) string { var failure *NativeCallFailure if errors.As(err, &failure) { return failure.Phase } return "unclassified" } // NativeCallConfirmedPreAnswerEnd requires a submitted native Dial followed // by a read of the same channel returning 404 after cleanup. A hangup request, // timeout, or failed status read alone is not termination evidence. func NativeCallConfirmedPreAnswerEnd(err error) bool { var failure *NativeCallFailure return errors.As(err, &failure) && failure.Phase == "answer_wait" && failure.ConfirmedEnd } // NativeCallNeverSubmitted is true only when no ARI Dial request was sent. // A failed Dial request is uncertain even if the subsequent cleanup succeeded. func NativeCallNeverSubmitted(err error) bool { switch NativeCallPhase(err) { case "validate", "ari_open", "rtp_listen", "subscribe", "create", "caller_id", "sip_call_id", "sip_bind": return true default: return false } } // NativeCallHTTPStatus extracts a numeric ARI HTTP status without retaining // a provider response body, credential or request URL in diagnostic logs. func NativeCallHTTPStatus(err error) int { var response native.RequestError for err != nil { if errors.As(err, &response) { return response.Code() } // Channel.Data wraps the HTTP error with a Cause() wrapper in ari/v5. var caused interface{ Cause() error } if !errors.As(err, &caused) { break } err = caused.Cause() } return 0 } // NativeCall owns one real ARI channel, bridge, ExternalMedia channel and RTP // socket; closing it cannot originate or replay a second call. type NativeCall struct { Media *media.RTPStream Context context.Context cancel context.CancelCauseFunc monitorDone chan struct{} destroyed atomic.Bool verifyEnd bool confirmedEnd bool client ari.Client subscription ari.Subscription outbound *ari.ChannelHandle external *ari.ChannelHandle bridge *ari.BridgeHandle closeOnce sync.Once closeErr error } // Originate uses the issued Cell-local ARI credential. The caller must have // already persisted the signed instruction and passed the host capture gate. func (l Loader) Originate(ctx context.Context, request NativeDial) (*NativeCall, error) { if _, err := approvedOriginateRequest(request.ExecutionID, request.TrunkID, request.DialedCallee, request.CallerID, request.AnswerTimeout); err != nil { return nil, &NativeCallFailure{Phase: "validate", Cause: err} } client, err := l.OpenARI() if err != nil { return nil, &NativeCallFailure{Phase: "ari_open", Cause: err} } return dialWithClient(ctx, client, request) } func dialWithClient(ctx context.Context, client ari.Client, request NativeDial) (_ *NativeCall, err error) { call := &NativeCall{client: client} defer func() { if err != nil { cleanupErr := call.Close() var failure *NativeCallFailure if errors.As(err, &failure) && failure.Phase == "answer_wait" && call.confirmedEnd { failure.ConfirmedEnd = true } err = errors.Join(err, cleanupErr) } }() if ctx == nil || client == nil { return nil, errors.New("native ARI call context and client required") } call.Context, call.cancel = context.WithCancelCause(ctx) originate, err := approvedOriginateRequest(request.ExecutionID, request.TrunkID, request.DialedCallee, request.CallerID, request.AnswerTimeout) if err != nil { return nil, &NativeCallFailure{Phase: "validate", Cause: err} } if request.BindSIPCall == nil { return nil, &NativeCallFailure{Phase: "validate", Cause: errors.New("HEP SIP call binding is required before native Dial")} } call.Media, err = media.ListenRTPWithFormat("127.0.0.1:0", request.MediaPayloadType, media.FormatSLIN16, 16000) if err != nil { return nil, &NativeCallFailure{Phase: "rtp_listen", Cause: err} } key := ari.NewKey(ari.ChannelKey, request.ExecutionID) call.subscription = client.Bus().Subscribe(key, "StasisStart", "ChannelStateChange", "StasisEnd", "ChannelHangupRequest", "ChannelDestroyed") if call.subscription == nil { return nil, &NativeCallFailure{Phase: "subscribe", Cause: errors.New("native ARI channel subscription unavailable before origination")} } // Create joins Stasis without sending SIP. The Asterisk channel's own // PJSIP Call-ID is available before Dial, even for immediate SIP 480. call.outbound, err = client.Channel().Create(nil, ari.ChannelCreateRequest{ ChannelID: originate.ChannelID, Endpoint: originate.Endpoint, App: originate.App, }) if err != nil { call.outbound = client.Channel().Get(key) // HTTP outcome may be lost; no Dial was submitted. return nil, &NativeCallFailure{Phase: "create", Cause: err} } if err := call.outbound.SetVariable("CALLERID(num)", originate.CallerID); err != nil { return nil, &NativeCallFailure{Phase: "caller_id", Cause: err} } callID, err := call.outbound.GetVariable("CHANNEL(pjsip,call-id)") if err != nil || strings.TrimSpace(callID) == "" { return nil, &NativeCallFailure{Phase: "sip_call_id", Cause: errors.Join(err, errors.New("Asterisk did not provide a pre-Dial SIP Call-ID"))} } if err := request.BindSIPCall(request.ExecutionID, callID); err != nil { return nil, &NativeCallFailure{Phase: "sip_bind", Cause: err} } if err := call.outbound.Dial("", time.Duration(originate.Timeout)*time.Second); err != nil { // The Dial HTTP reply may be lost after SIP was sent. Never Dial twice. return nil, &NativeCallFailure{Phase: "dial", Cause: err} } answerCtx, cancel := context.WithTimeout(ctx, request.AnswerTimeout) defer cancel() if err := awaitDialUp(answerCtx, call.subscription, request.ExecutionID); err != nil { call.verifyEnd = true // Created channel has no ExternalMedia or bridge yet. return nil, &NativeCallFailure{Phase: "answer_wait", Cause: err} } bridgeKey := ari.NewKey(ari.BridgeKey, request.ExecutionID+"-bridge") call.bridge, err = client.Bridge().Create(bridgeKey, "mixing", "go-sip-agent") if err != nil { // A lost reply does not prove that the bridge was not created. call.bridge = client.Bridge().Get(bridgeKey) return nil, &NativeCallFailure{Phase: "bridge_create", Cause: err} } if err := call.bridge.AddChannel(call.outbound.ID()); err != nil { return nil, &NativeCallFailure{Phase: "bridge_add_outbound", Cause: err} } mediaKey := ari.NewKey(ari.ChannelKey, request.ExecutionID+"-media") call.external, err = client.Channel().ExternalMedia(mediaKey, ari.ExternalMediaOptions{ App: "go-sip-agent", ExternalHost: net.JoinHostPort("127.0.0.1", strconv.Itoa(call.Media.LocalAddr().(*net.UDPAddr).Port)), Format: "slin16", Encapsulation: "rtp", Transport: "udp", ConnectionType: "client", Direction: "both", }) if err != nil { // Use the same identity to stop an ExternalMedia channel created // before the HTTP outcome became unknown; never create another. call.external = client.Channel().Get(mediaKey) return nil, &NativeCallFailure{Phase: "external_media_create", Cause: err} } address, err := call.external.GetVariable("UNICASTRTP_LOCAL_ADDRESS") if err != nil { return nil, &NativeCallFailure{Phase: "rtp_peer_address", Cause: err} } port, err := call.external.GetVariable("UNICASTRTP_LOCAL_PORT") if err != nil { return nil, &NativeCallFailure{Phase: "rtp_peer_port", Cause: err} } peer, err := netip.ParseAddr(address) if err != nil || !peer.IsLoopback() { return nil, &NativeCallFailure{Phase: "rtp_peer_validation", Cause: errors.New("Agent ARI RTP peer is not a local Cell address")} } if err := call.Media.SetPeer(net.JoinHostPort(address, port)); err != nil { return nil, &NativeCallFailure{Phase: "rtp_peer_config", Cause: err} } if err := call.bridge.AddChannel(call.external.ID()); err != nil { return nil, &NativeCallFailure{Phase: "bridge_add_media", Cause: err} } call.monitorDone = make(chan struct{}) go call.watch(request.ExecutionID) return call, nil } func (c *NativeCall) watch(channelID string) { defer close(c.monitorDone) for { select { case <-c.Context.Done(): return case event, open := <-c.subscription.Events(): if !open || event == nil { c.cancel(errors.New("native ARI event subscription lost during call")) return } if !eventBelongsToChannel(event, channelID) { continue } switch event.GetType() { case "ChannelDestroyed": if ended, ok := event.(*ari.ChannelDestroyed); ok { c.destroyed.Store(true) c.cancel(fmt.Errorf("native channel ended: cause=%d", ended.Cause)) } else { c.cancel(errors.New("native channel ended: invalid destroy event")) } return case "ChannelHangupRequest", "StasisEnd": c.cancel(fmt.Errorf("native channel ended: event=%s", event.GetType())) return } } } } // Close releases each resource once. Errors remain visible to the caller; a // failed Hangup never silently turns an uncertain call into a success. func (c *NativeCall) Close() error { if c == nil { return nil } c.closeOnce.Do(func() { if c.cancel != nil { c.cancel(context.Canceled) } if c.subscription != nil { c.subscription.Cancel() } if c.external != nil { c.closeErr = errors.Join(c.closeErr, c.external.Hangup()) } if c.outbound != nil && !c.destroyed.Load() { c.closeErr = errors.Join(c.closeErr, c.outbound.Hangup()) } if c.verifyEnd && c.outbound != nil { _, err := c.outbound.Data() if NativeCallHTTPStatus(err) == 404 { c.confirmedEnd = true } else if err != nil { c.closeErr = errors.Join(c.closeErr, fmt.Errorf("native outbound channel end confirmation unavailable: %w", err)) } } if c.bridge != nil { c.closeErr = errors.Join(c.closeErr, c.bridge.Delete()) } if c.Media != nil { c.closeErr = errors.Join(c.closeErr, c.Media.Close()) } if c.client != nil { c.client.Close() } if c.monitorDone != nil { <-c.monitorDone } }) return c.closeErr }