Dev

pkg/go.mod

@@ -3,27 +3,45 @@ module github.com/samims/hcaas/pkg
 go 1.24.4
 
 require (
+	github.com/IBM/sarama v1.45.2
 	go.opentelemetry.io/otel v1.37.0
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.37.0
 	go.opentelemetry.io/otel/sdk v1.37.0
-	google.golang.org/grpc v1.73.0
+	go.opentelemetry.io/otel/trace v1.37.0
 )
 
 require (
 	github.com/cenkalti/backoff/v5 v5.0.2 // indirect
+	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/eapache/go-resiliency v1.7.0 // indirect
+	github.com/eapache/go-xerial-snappy v0.0.0-20230731223053-c322873962e3 // indirect
+	github.com/eapache/queue v1.1.0 // indirect
 	github.com/go-logr/logr v1.4.3 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
+	github.com/golang/snappy v0.0.4 // indirect
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.1 // indirect
+	github.com/hashicorp/errwrap v1.0.0 // indirect
+	github.com/hashicorp/go-multierror v1.1.1 // indirect
+	github.com/hashicorp/go-uuid v1.0.3 // indirect
+	github.com/jcmturner/aescts/v2 v2.0.0 // indirect
+	github.com/jcmturner/dnsutils/v2 v2.0.0 // indirect
+	github.com/jcmturner/gofork v1.7.6 // indirect
+	github.com/jcmturner/gokrb5/v8 v8.4.4 // indirect
+	github.com/jcmturner/rpc/v2 v2.0.3 // indirect
+	github.com/klauspost/compress v1.18.0 // indirect
+	github.com/pierrec/lz4/v4 v4.1.22 // indirect
+	github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475 // indirect
 	go.opentelemetry.io/auto/sdk v1.1.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.37.0 // indirect
 	go.opentelemetry.io/otel/metric v1.37.0 // indirect
-	go.opentelemetry.io/otel/trace v1.37.0 // indirect
 	go.opentelemetry.io/proto/otlp v1.7.0 // indirect
+	golang.org/x/crypto v0.39.0 // indirect
 	golang.org/x/net v0.41.0 // indirect
 	golang.org/x/sys v0.33.0 // indirect
 	golang.org/x/text v0.26.0 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20250603155806-513f23925822 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20250603155806-513f23925822 // indirect
+	google.golang.org/grpc v1.73.0 // indirect
 	google.golang.org/protobuf v1.36.6 // indirect
 )

pkg/tracing/tracer.go

@@ -79,6 +79,10 @@ func (t *Tracer) AddAttributes(span trace.Span, attrs ...attribute.KeyValue) {
 	span.SetAttributes(attrs...)
 }
 
+func (t *Tracer) AddEvent(span trace.Span, name string, attrs ...attribute.KeyValue) {
+	span.AddEvent(name, trace.WithAttributes(attrs...))
+}
+
 // AddGoogleCloudAttributes adds Google Cloud specific attributes
 func (t *Tracer) AddGoogleCloudAttributes(span trace.Span, projectID, region, zone string) {
 	span.SetAttributes(
@@ -131,3 +135,11 @@ func (t *Tracer) AddKafkaAttributes(span trace.Span, topic, operation string, pa
 func GetTracer(name string) trace.Tracer {
 	return otel.Tracer(name)
 }
+
+// GetTraceID extracts the trace ID from the context
+func (t *Tracer) GetTraceID(ctx context.Context) string {
+	if span := trace.SpanFromContext(ctx); span.SpanContext().IsValid() {
+		return span.SpanContext().TraceID().String()
+	}
+	return ""
+}

services/auth/internal/errors/errors.go

@@ -3,11 +3,12 @@ package errors
 import "errors"
 
 var (
+	ErrInvalidEmail    = errors.New("invalid email")
+	ErrInvalidInput    = errors.New("invalid input")
+	ErrConflict        = errors.New("conflict")
 	ErrInternal        = errors.New("internal error")
-	ErrNotFound        = errors.New("resource not found")
 	ErrUnauthorized    = errors.New("unauthorized")
-	ErrConflict        = errors.New("resource already exists")
-	ErrTokenGeneration = errors.New("token generation failed ")
-	ErrInvalidEmail    = errors.New("invalid Email")
-	ErrInvalidInput    = errors.New("invalid input")
+	ErrTokenGeneration = errors.New("token generation failed")
+	ErrTooManyAttempts = errors.New("too many login attempts, account locked temporarily")
+	ErrNotFound        = errors.New("not found")
 )

services/auth/internal/service/auth_service.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"log/slog"
 	"regexp"
+	"time"
 
 	"golang.org/x/crypto/bcrypt"
 
@@ -26,23 +27,48 @@ type authService struct {
 	store    storage.UserStorage
 	logger   *slog.Logger
 	tokenSvc TokenService
+	// Add rate limiting map or store here if needed
+	loginAttempts   map[string]int
+	lockoutTime     map[string]time.Time
+	lockoutDuration time.Duration
+	maxAttempts     int
 }
 
 func NewAuthService(store storage.UserStorage, logger *slog.Logger, tokenSvc TokenService) AuthService {
 	l := logger.With("layer", "service", "component", "authService")
-	return &authService{store: store, logger: l, tokenSvc: tokenSvc}
+	return &authService{
+		store:           store,
+		logger:          l,
+		tokenSvc:        tokenSvc,
+		loginAttempts:   make(map[string]int),
+		lockoutTime:     make(map[string]time.Time),
+		lockoutDuration: 15 * time.Minute,
+		maxAttempts:     5,
+	}
 }
 
 func (s *authService) Register(ctx context.Context, email, password string) (*model.User, error) {
 	s.logger.Info("Register called", slog.String("email", email))
 
-	if !regexp.MustCompile(`^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$`).MatchString(email) {
+	if !regexp.MustCompile(`^[a-zA-Z0-9._%+\-]+@[a-zA-Z0-9.\-]+\.[a-zA-Z]{2,}$`).MatchString(email) {
 		s.logger.Error("Invalid email")
 		return nil, appErr.ErrInvalidEmail
 	}
 
-	if len(password) == 0 {
-		s.logger.Error("Invalid password")
+	if len(password) < 8 {
+		s.logger.Error("Password too short")
+		return nil, appErr.ErrInvalidInput
+	}
+
+	// Enforce password complexity: at least one uppercase, one lowercase, one digit, one special char
+	var (
+		hasUpper   = regexp.MustCompile(`[A-Z]`).MatchString
+		hasLower   = regexp.MustCompile(`[a-z]`).MatchString
+		hasDigit   = regexp.MustCompile(`[0-9]`).MatchString
+		hasSpecial = regexp.MustCompile(`[\W_]`).MatchString
+	)
+	if !hasUpper(password) || !hasLower(password) || !hasDigit(password) || !hasSpecial(password) {
+		s.logger.Error("Password does not meet complexity requirements")
 		return nil, appErr.ErrInvalidInput
 	}
 
@@ -69,6 +95,18 @@ func (s *authService) Register(ctx context.Context, email, password string) (*mo
 func (s *authService) Login(ctx context.Context, email, password string) (*model.User, string, error) {
 	s.logger.Info("Login called", slog.String("email", email))
 
+	// Check if user is locked out
+	if lockoutUntil, locked := s.lockoutTime[email]; locked {
+		if time.Now().Before(lockoutUntil) {
+			s.logger.Warn("User account locked due to too many failed login attempts", slog.String("email", email))
+			return nil, "", appErr.ErrTooManyAttempts
+		} else {
+			// Lockout expired, reset
+			delete(s.lockoutTime, email)
+			s.loginAttempts[email] = 0
+		}
+	}
+
 	user, err := s.store.GetUserByEmail(ctx, email)
 	if err != nil {
 		if errors.Is(err, pgx.ErrNoRows) {
@@ -82,9 +120,19 @@ func (s *authService) Login(ctx context.Context, email, password string) (*model
 
 	// Compare the provided password with the stored hashed password
 	if err := bcrypt.CompareHashAndPassword([]byte(user.Password), []byte(password)); err != nil {
+		s.loginAttempts[email]++
+		if s.loginAttempts[email] >= s.maxAttempts {
+			s.lockoutTime[email] = time.Now().Add(s.lockoutDuration)
+			s.logger.Warn("User account locked due to too many failed login attempts", slog.String("email", email))
+			return nil, "", appErr.ErrTooManyAttempts
+		}
 		s.logger.Warn("Invalid password", slog.String("email", email))
 		return nil, "", appErr.ErrUnauthorized
 	}
+
+	// Reset login attempts on successful login
+	s.loginAttempts[email] = 0
+
 	token, err := s.tokenSvc.GenerateToken(user)
 	if err != nil {
 		s.logger.Error("Token generation failed ", slog.String("email", email))

services/auth/internal/service/auth_service_test.go

@@ -52,7 +52,7 @@ func Test_authService_Register(t *testing.T) {
 			args: args{
 				ctx:      context.Background(),
 				email:    "[email protected]",
-				password: "password123",
+				password: "Password@123",
 			},
 			want: &model.User{
 				Email: "[email protected]",
@@ -74,7 +74,7 @@ func Test_authService_Register(t *testing.T) {
 			args: args{
 				ctx:      context.Background(),
 				email:    "[email protected]",
-				password: "password123",
+				password: "Password@123",
 			},
 			want:    nil,
 			wantErr: true,
@@ -92,7 +92,7 @@ func Test_authService_Register(t *testing.T) {
 			args: args{
 				ctx:      context.Background(),
 				email:    "",
-				password: "password123",
+				password: "Password@123",
 			},
 			want:    nil,
 			wantErr: true,
@@ -110,7 +110,7 @@ func Test_authService_Register(t *testing.T) {
 			args: args{
 				ctx:      context.Background(),
 				email:    "test1.example.com",
-				password: "password123",
+				password: "Password@123",
 			},
 			want:    nil,
 			wantErr: true,

services/auth/internal/service/token_service.go

@@ -37,13 +37,19 @@ func (s *jwtService) GenerateToken(user *model.User) (string, error) {
 		"email": user.Email,
 		"exp":   time.Now().Add(s.expiryTime).Unix(),
 		"iat":   time.Now().Unix(),
+		"nbf":   time.Now().Unix(), // Not valid before now
 	}
 	token := jwt.NewWithClaims(jwt.SigningMethodHS256, claims)
 	return token.SignedString([]byte(s.secret))
 }
 
 func (s *jwtService) ValidateToken(tokenStr string) (string, string, error) {
 	token, err := jwt.Parse(tokenStr, func(token *jwt.Token) (any, error) {
+		// Validate the signing method to prevent algorithm confuses attack
+		if _, ok := token.Method.(*jwt.SigningMethodHMAC); !ok {
+			s.logger.Error("Unexpected signing method", slog.String("method", token.Header["alg"].(string)))
+			return nil, jwt.ErrSignatureInvalid
+		}
 		return []byte(s.secret), nil
 	})
 
@@ -63,6 +69,29 @@ func (s *jwtService) ValidateToken(tokenStr string) (string, string, error) {
 		return "", "", jwt.ErrTokenMalformed
 	}
 	email, ok := claims["email"].(string)
+	if !ok {
+		s.logger.Error("Invalid email claim", slog.String("email", email))
+		return "", "", jwt.ErrTokenMalformed
+	}
+
+	// validate time based claims
+	now := time.Now().Unix()
+
+	// check expiry
+	if exp, ok := claims["exp"].(float64); ok {
+		if int64(exp) < now {
+			s.logger.Error("Token expired", slog.Int64("exp", int64(exp)), slog.Int64("now", now))
+			return "", "", jwt.ErrTokenExpired
+		}
+	}
+
+	// check not before
+	if nbf, ok := claims["nbf"].(float64); ok {
+		if int64(nbf) > now {
+			s.logger.Error("Token not valid yet", slog.Int64("nbf", int64(nbf)), slog.Int64("now", now))
+			return "", "", jwt.ErrTokenNotValidYet
+		}
+	}
 
 	return userID, email, nil
 }

services/auth/internal/storage/user_storage.go

@@ -2,7 +2,6 @@ package storage
 
 import (
 	"context"
-	"fmt"
 	"time"
 
 	"github.com/jackc/pgx/v5/pgxpool"
@@ -60,8 +59,6 @@ func (s *userStorage) GetUserByEmail(ctx context.Context, email string) (*model.
 	if err := row.Scan(&user.ID, &user.Email, &user.Password, &user.CreatedAt); err != nil {
 		return nil, err
 	}
-	fmt.Println(user)
-
 	return &user, nil
 }
 

services/url/cmd/url/main.go

@@ -111,7 +111,8 @@ func main() {
 	notificationProducer.Start(ctx)
 
 	httpClient := &http.Client{Timeout: 5 * time.Second}
-	chkr := checker.NewURLChecker(urlSvc, l, httpClient, 1*time.Minute, notificationProducer, tracer)
+	concurrencyLimit := 10
+	chkr := checker.NewURLChecker(urlSvc, l, httpClient, 1*time.Minute, notificationProducer, tracer, concurrencyLimit)
 	go chkr.Start(ctx)
 
 	urlHandler := handler.NewURLHandler(urlSvc, l)

services/url/internal/checker/checker.go

@@ -28,6 +28,8 @@ type URLChecker struct {
 	interval             time.Duration
 	notificationProducer kafka.NotificationProducer
 	tracer               *tracing.Tracer
+	concurrencyLimit     int
+	httpTimeOut          time.Duration
 }
 
 func NewURLChecker(
@@ -37,6 +39,7 @@ func NewURLChecker(
 	interval time.Duration,
 	producer kafka.NotificationProducer,
 	tracer *tracing.Tracer,
+	concurrencyLimit int,
 ) *URLChecker {
 	if producer == nil {
 		// This panic indicates a serious configuration error that should be caught
@@ -52,6 +55,7 @@ func NewURLChecker(
 		interval:             interval,
 		notificationProducer: producer,
 		tracer:               tracer,
+		concurrencyLimit:     concurrencyLimit,
 	}
 }
 
@@ -79,12 +83,13 @@ func (uc *URLChecker) CheckAllURLs(ctx context.Context) {
 	urls, err := uc.svc.GetAll(ctx)
 	if err != nil {
 		uc.logger.Error("Failed to fetch URLs", slog.Any("error", err))
-		span.RecordError(err)
+		uc.tracer.RecordError(span, err)
 		return
 	}
 
 	var wg sync.WaitGroup
-	sem := make(chan struct{}, 10) // Limit to 10 concurrent checks
+	// Use the configurable concurrency limit for the semaphore.
+	sem := make(chan struct{}, uc.concurrencyLimit)
 
 	for _, url := range urls {
 		wg.Add(1)
@@ -114,7 +119,7 @@ func (uc *URLChecker) CheckAllURLs(ctx context.Context) {
 					slog.String("status", status),
 					slog.Any("error", err),
 				)
-				span.RecordError(err)
+				uc.tracer.RecordError(span, err)
 			} else {
 				uc.logger.Info("URL status updated",
 					slog.String("urlID", url.ID),
@@ -135,7 +140,7 @@ func (uc *URLChecker) CheckAllURLs(ctx context.Context) {
 						uc.logger.Error("Failed to publish notification",
 							slog.String("url_id", url.ID),
 							slog.Any("error", err))
-						span.RecordError(err)
+						uc.tracer.RecordError(span, err)
 					}
 				}
 			}