fix(sbi): enforce inbound OAuth2 on NEF service routes

config/nefcfg.yaml

@@ -14,8 +14,10 @@ configuration:
   nrfUri: http://127.0.0.10:8000 # A valid URI of NRF
   nrfCertPem: cert/nrf.pem # NRF Certificate
   serviceList: # the SBI services provided by this NEF
-    - serviceName: nnef-pfdmanagement # Nnef_PFDManagement Service
-    - serviceName: nnef-oam # OAM service
+    - serviceName: nnef-pfdmanagement  # Nnef_PFDManagement Service (also mounts /3gpp-pfd-management)
+    - serviceName: nnef-oam            # OAM service
+    - serviceName: 3gpp-traffic-influence # AF-facing Traffic Influence API
+    - serviceName: nnef-callback       # Inbound SMF event notification callback
 
 logger: # log output setting
   enable: true # true or false

internal/context/nef_context.go

@@ -15,6 +15,13 @@ type nef interface {
 	Config() *factory.Config
 }
 
+// NFContext is the interface used by middleware to perform inbound OAuth2 token checks.
+type NFContext interface {
+	AuthorizationCheck(token string, serviceName models.ServiceName) error
+}
+
+var _ NFContext = &NefContext{}
+
 type NefContext struct {
 	nef
 
@@ -159,3 +166,14 @@ func (c *NefContext) GetTokenCtx(serviceName models.ServiceName, targetNF models
 	return oauth.GetTokenCtx(models.NrfNfManagementNfType_NEF, targetNF,
 		c.nfInstID, c.Config().NrfUri(), string(serviceName))
 }
+
+// AuthorizationCheck validates the inbound OAuth2 bearer token against serviceName.
+// When OAuth2 is disabled it returns nil immediately (pass-through for dev/test).
+func (c *NefContext) AuthorizationCheck(token string, serviceName models.ServiceName) error {
+	if !c.OAuth2Required {
+		logger.CtxLog.Debugf("NefContext::AuthorizationCheck: OAuth2 not required")
+		return nil
+	}
+	logger.CtxLog.Debugf("NefContext::AuthorizationCheck: token[%s] serviceName[%s]", token, serviceName)
+	return oauth.VerifyOAuth(token, string(serviceName), c.Config().NrfCertPem())
+}

internal/logger/logger.go

@@ -17,6 +17,7 @@ var (
 	SBILog       *logrus.Entry
 	ConsumerLog  *logrus.Entry
 	ProcessorLog *logrus.Entry
+	UtilLog      *logrus.Entry
 	TrafInfluLog *logrus.Entry
 	PFDManageLog *logrus.Entry
 	PFDFLog      *logrus.Entry
@@ -48,6 +49,7 @@ func init() {
 	SBILog = NfLog.WithField(logger_util.FieldCategory, "SBI")
 	ConsumerLog = NfLog.WithField(logger_util.FieldCategory, "Consumer")
 	ProcessorLog = NfLog.WithField(logger_util.FieldCategory, "Proc")
+	UtilLog = NfLog.WithField(logger_util.FieldCategory, "Util")
 	TrafInfluLog = NfLog.WithField(logger_util.FieldCategory, "TraffInfl")
 	PFDManageLog = NfLog.WithField(logger_util.FieldCategory, "PFDMng")
 	PFDFLog = NfLog.WithField(logger_util.FieldCategory, "PFDF")

internal/sbi/processor/callback.go

@@ -1,15 +1,22 @@
 package processor
 
 import (
+	"bytes"
+	"context"
+	"fmt"
+	"io"
 	"net/http"
 
 	"github.com/free5gc/nef/internal/logger"
 	"github.com/free5gc/openapi"
 	"github.com/free5gc/openapi/models"
 	"github.com/free5gc/util/metrics/sbi"
 	"github.com/gin-gonic/gin"
+	"golang.org/x/oauth2"
 )
 
+var afCallbackHTTPClient = &http.Client{}
+
 func (p *Processor) SmfNotification(
 	c *gin.Context,
 	eeNotif *models.NsmfEventExposureNotification,
@@ -25,9 +32,100 @@ func (p *Processor) SmfNotification(
 	}
 
 	af.Mu.RLock()
-	defer af.Mu.RUnlock()
+	notifDestination := ""
+	if sub.TiSub != nil {
+		notifDestination = sub.TiSub.NotificationDestination
+	}
+	af.Mu.RUnlock()
+
+	if notifDestination == "" {
+		pd := openapi.ProblemDetailsSystemFailure("AF notification destination is empty")
+		c.Set(sbi.IN_PB_DETAILS_CTX_STR, pd.Cause)
+		c.JSON(http.StatusInternalServerError, pd)
+		return
+	}
+
+	afCallbackTokenCtx, pd, err := p.Context().GetTokenCtx(
+		models.ServiceName("nnef-callback"), models.NrfNfManagementNfType_AF)
+	if err != nil {
+		logger.TrafInfluLog.Errorf("Get token for AF callback failed: %+v", pd)
+		failure := openapi.ProblemDetailsSystemFailure("get token for AF callback failed")
+		if pd != nil && pd.Cause != "" {
+			c.Set(sbi.IN_PB_DETAILS_CTX_STR, pd.Cause)
+		} else {
+			c.Set(sbi.IN_PB_DETAILS_CTX_STR, failure.Cause)
+		}
+		c.JSON(http.StatusBadGateway, failure)
+		return
+	}
+
+	if err := postSmfEventExposureNotificationToAf(notifDestination, eeNotif, afCallbackTokenCtx); err != nil {
+		logger.TrafInfluLog.Errorf("Forward SMF notification to AF failed: %v", err)
+		pd := openapi.ProblemDetailsSystemFailure(err.Error())
+		c.Set(sbi.IN_PB_DETAILS_CTX_STR, pd.Cause)
+		c.JSON(http.StatusBadGateway, pd)
+		return
+	}
+
+	c.Status(http.StatusNoContent)
+}
+
+func postSmfEventExposureNotificationToAf(
+	notifDestination string,
+	eeNotif *models.NsmfEventExposureNotification,
+	requestCtx context.Context,
+) error {
+	reqBody, err := openapi.Serialize(eeNotif, "application/json")
+	if err != nil {
+		return fmt.Errorf("serialize SMF notification failed: %w", err)
+	}
+	if requestCtx == nil {
+		requestCtx = context.Background()
+	}
+
+	httpReq, err := http.NewRequestWithContext(requestCtx, http.MethodPost, notifDestination, bytes.NewReader(reqBody))
+	if err != nil {
+		return fmt.Errorf("create AF callback request failed: %w", err)
+	}
+	httpReq.Header.Set("Content-Type", "application/json")
+	if err = bindOAuthTokenToRequest(httpReq, requestCtx); err != nil {
+		return fmt.Errorf("bind OAuth2 token for AF callback failed: %w", err)
+	}
+
+	httpRsp, err := afCallbackHTTPClient.Do(httpReq)
+	if err != nil {
+		return fmt.Errorf("send AF callback failed: %w", err)
+	}
+	defer func() {
+		if _, copyErr := io.Copy(io.Discard, httpRsp.Body); copyErr != nil {
+			logger.TrafInfluLog.Warnf("drain AF callback response body failed: %v", copyErr)
+		}
+		if closeErr := httpRsp.Body.Close(); closeErr != nil {
+			logger.TrafInfluLog.Warnf("close AF callback response body failed: %v", closeErr)
+		}
+	}()
 
-	// TODO: Notify AF
+	if httpRsp.StatusCode < http.StatusOK || httpRsp.StatusCode >= http.StatusMultipleChoices {
+		return fmt.Errorf("AF callback returned status code %d", httpRsp.StatusCode)
+	}
+
+	return nil
+}
+
+func bindOAuthTokenToRequest(req *http.Request, requestCtx context.Context) error {
+	if requestCtx == nil {
+		return nil
+	}
+
+	tok, ok := requestCtx.Value(openapi.ContextOAuth2).(oauth2.TokenSource)
+	if !ok {
+		return nil
+	}
 
-	c.JSON(http.StatusOK, nil)
+	latestToken, err := tok.Token()
+	if err != nil {
+		return err
+	}
+	latestToken.SetAuthHeader(req)
+	return nil
 }

internal/sbi/processor/callback_test.go

@@ -0,0 +1,97 @@
+package processor
+
+import (
+	"context"
+	"io"
+	"net/http"
+	"strings"
+	"testing"
+
+	"github.com/free5gc/openapi"
+	"github.com/free5gc/openapi/models"
+	"golang.org/x/oauth2"
+)
+
+type roundTripFunc func(*http.Request) (*http.Response, error)
+
+func (f roundTripFunc) RoundTrip(req *http.Request) (*http.Response, error) {
+	return f(req)
+}
+
+func TestBindOAuthTokenToRequest(t *testing.T) {
+	t.Run("context without token source", func(t *testing.T) {
+		req, err := http.NewRequestWithContext(context.Background(), http.MethodPost, "http://example.com", nil)
+		if err != nil {
+			t.Fatalf("create request failed: %v", err)
+		}
+
+		err = bindOAuthTokenToRequest(req, context.TODO())
+		if err != nil {
+			t.Fatalf("bind token failed: %v", err)
+		}
+		if got := req.Header.Get("Authorization"); got != "" {
+			t.Fatalf("unexpected authorization header: %q", got)
+		}
+	})
+
+	t.Run("context with token source", func(t *testing.T) {
+		req, err := http.NewRequestWithContext(context.Background(), http.MethodPost, "http://example.com", nil)
+		if err != nil {
+			t.Fatalf("create request failed: %v", err)
+		}
+
+		tok := oauth2.StaticTokenSource(&oauth2.Token{AccessToken: "abc123", TokenType: "Bearer"})
+		tokenCtx := context.WithValue(context.Background(), openapi.ContextOAuth2, tok)
+
+		err = bindOAuthTokenToRequest(req, tokenCtx)
+		if err != nil {
+			t.Fatalf("bind token failed: %v", err)
+		}
+		if got := req.Header.Get("Authorization"); got != "Bearer abc123" {
+			t.Fatalf("authorization header = %q, want %q", got, "Bearer abc123")
+		}
+	})
+}
+
+func TestPostSmfEventExposureNotificationToAfWithToken(t *testing.T) {
+	originalClient := afCallbackHTTPClient
+	t.Cleanup(func() { afCallbackHTTPClient = originalClient })
+
+	afCallbackHTTPClient = &http.Client{Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) {
+		if got := req.Header.Get("Authorization"); got != "Bearer token-for-af" {
+			t.Fatalf("authorization header = %q, want %q", got, "Bearer token-for-af")
+		}
+		return &http.Response{
+			StatusCode: http.StatusNoContent,
+			Body:       io.NopCloser(strings.NewReader("")),
+			Header:     make(http.Header),
+		}, nil
+	})}
+
+	tok := oauth2.StaticTokenSource(&oauth2.Token{AccessToken: "token-for-af", TokenType: "Bearer"})
+	tokenCtx := context.WithValue(context.Background(), openapi.ContextOAuth2, tok)
+
+	eeNotif := &models.NsmfEventExposureNotification{NotifId: "notif-1"}
+	if err := postSmfEventExposureNotificationToAf("http://af.example.com/notify", eeNotif, tokenCtx); err != nil {
+		t.Fatalf("post callback failed: %v", err)
+	}
+}
+
+func TestPostSmfEventExposureNotificationToAfNon2xx(t *testing.T) {
+	originalClient := afCallbackHTTPClient
+	t.Cleanup(func() { afCallbackHTTPClient = originalClient })
+
+	afCallbackHTTPClient = &http.Client{Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) {
+		return &http.Response{
+			StatusCode: http.StatusForbidden,
+			Body:       io.NopCloser(strings.NewReader("forbidden")),
+			Header:     make(http.Header),
+		}, nil
+	})}
+
+	eeNotif := &models.NsmfEventExposureNotification{NotifId: "notif-2"}
+	err := postSmfEventExposureNotificationToAf("http://af.example.com/notify", eeNotif, context.TODO())
+	if err == nil {
+		t.Fatal("expected error when AF callback returns non-2xx")
+	}
+}

internal/sbi/processor/pfd.go

@@ -67,8 +67,6 @@ func (p *Processor) PostPFDManagementTransactions(
 ) {
 	logger.PFDManageLog.Infof("PostPFDManagementTransactions - scsAsID[%s]", scsAsID)
 
-	// TODO: Authorize the AF
-
 	nefCtx := p.Context()
 	if pd := validatePfdManagement(scsAsID, "-1", pfdMng, nefCtx); pd != nil {
 		if pd.Status == http.StatusInternalServerError {
@@ -234,8 +232,6 @@ func (p *Processor) PutIndividualPFDManagementTransaction(
 	logger.PFDManageLog.Infof("PutIndividualPFDManagementTransaction - scsAsID[%s], transID[%s]",
 		scsAsID, transID)
 
-	// TODO: Authorize the AF
-
 	nefCtx := p.Context()
 	if pd := validatePfdManagement(scsAsID, transID, pfdMng, nefCtx); pd != nil {
 		if pd.Status == http.StatusInternalServerError {
@@ -502,8 +498,6 @@ func (p *Processor) PutIndividualApplicationPFDManagement(
 	logger.PFDManageLog.Infof("PutIndividualApplicationPFDManagement - scsAsID[%s], transID[%s], appID[%s]",
 		scsAsID, transID, appID)
 
-	// TODO: Authorize the AF
-
 	nefCtx := p.Context()
 	af := nefCtx.GetAf(scsAsID)
 	if af == nil {
@@ -568,8 +562,6 @@ func (p *Processor) PatchIndividualApplicationPFDManagement(
 	logger.PFDManageLog.Infof("PatchIndividualApplicationPFDManagement - scsAsID[%s], transID[%s], appID[%s]",
 		scsAsID, transID, appID)
 
-	// TODO: Authorize the AF
-
 	nefCtx := p.Context()
 	af := nefCtx.GetAf(scsAsID)
 	if af == nil {

internal/sbi/processor/ti.go

@@ -2,6 +2,7 @@ package processor
 
 import (
 	"net/http"
+	"net/url"
 
 	"github.com/free5gc/nef/internal/logger"
 	"github.com/free5gc/nef/pkg/factory"
@@ -416,6 +417,17 @@ func (p *Processor) DeleteIndividualTrafficInfluenceSubscription(
 func validateTrafficInfluenceData(
 	tiSub *models.NefTrafficInfluSub,
 ) *models.ProblemDetails {
+	if tiSub.NotificationDestination == "" {
+		pd := openapi.ProblemDetailsMalformedReqSyntax("Missing notificationDestination")
+		return pd
+	}
+
+	parsedNotificationDestination, err := url.ParseRequestURI(tiSub.NotificationDestination)
+	if err != nil || parsedNotificationDestination.Scheme == "" || parsedNotificationDestination.Host == "" {
+		pd := openapi.ProblemDetailsMalformedReqSyntax("Invalid notificationDestination")
+		return pd
+	}
+
 	// TS29.522: One of "afAppId", "trafficFilters" or "ethTrafficFilters" shall be included.
 	if tiSub.AfAppId == "" &&
 		len(tiSub.TrafficFilters) == 0 &&