DEV Community: Tim Nelles

A small Script to Detect Sha1-Hulud 2.0 affected Packages in NPM Projects

Tim Nelles — Tue, 25 Nov 2025 16:28:42 +0000

As of November 25th, 2025, the Shai Hulud 2 supply-chain incident is still in the process of being documented. To help developers verify their own projects, I created a small Bash script that checks installed package versions against the list published by Wiz Security.

GitHub: https://github.com/opctim/shai-hulud-2-check

What the script does

The script walks through your project directory, scanning both package-lock.json, pnpm-lock.yaml and yarn.lock files.

It extracts all installed package versions and checks them against the Shai Hulud 2 indicator list.

You can also point it to a custom CSV file via an environment variable if you prefer offline or internal data sources.

If any compromised version is found, the script exits with a non-zero status, making it easy to integrate into CI pipelines.

Contribute / Improve / Extend

I’m not looking for attention; I just hope the tool is useful.

If you find anything that needs improvement: feel free to contribute.

The Coffee-Cam: How Our Office Coffee Machine Learned to Talk

Tim Nelles — Fri, 25 Apr 2025 16:04:14 +0000

It all started — like many great ideas — with coffee. Or rather, the lack of it.

In our office, we kept running into the same annoyance: you’d walk up to the coffee machine, full of caffeine dreams, only to find it mid-cleaning cycle. 30 minutes of nothing but frustration.

There had to be a better way.

The Idea That Was Born at the Coffee Machine

One day, someone joked:

“Wouldn’t it be nice if we could see from our desks whether the machine is cleaning or not?”

Sadly, our coffee machine doesn’t come with an API - even though it’s a Pro-User product. But we had a spare Raspberry Pi and an old webcam lying around. And just like that, a tiny coffee-fueled hackathon began.

Step 1: Camera + Motion

We mounted the webcam on the coffee machine and installed Motion on the Raspberry Pi. It gave us a simple web interface that streamed frames from the camera — perfect for keeping an eye on the machine’s display.

From there, it was easy to tell manually whether the “Cleaning” box was on screen or not. Go to your browser, type the Raspberry Pi’s IP and have a look at the webcam.

Good first step. But having to check the browser every time? Still not ideal.

So, back to the drawing board…

Step 2: Image Classification with TensorFlow

We needed automation.

The camera position is fixed, thus it should give us a great playground to differentiate between images.

So I fired up TensorFlow and started training a tiny neural network. I collected screenshots from the camera feed — some with the cleaning screen visible, some without.

With just a few dozen examples per category, the model got surprisingly reliable at classifying whether the machine was in cleaning mode or ready for action.

A small excerpt of what i had to cope with:

Step 3: A Slackbot With Attitude

The last missing piece: notifications.

As the coffee machine is the hardest worker in most offices, plain messages would’ve been boring. So the bot got a bit of personality.

I used ChatGPT to generate a few, here are the ones that I integrated:

messages = [
    "🧼 I'm at the spa. It's called luxury. For you: cleaning mode.",
    "😶‍🌫️ Too much of your love is stuck on me. Let me scrub that off.",
    "🛁 Spa day for me. You guys take a break.",
    "😐 No, I'm not broken. I'm busy with a bubble bath.",
    "🧽 I'm cleaning myself. Why? Because you treat me like an animal.",
    "💤 Cleaning in progress. Please don't knock, I'm sensitive.",
    "🙃 Hygiene is important. That's what my therapist told me.",
    "🚿 I'm showering. You should try it too.",
    "🤖 Maintenance ongoing. Mood? Somewhere between resigned and soapy.",
    "🫣 I'm getting rid of your coffee sins. Don't look."
] if is_positive else [
    "☕ Ready for the next coffee. Wishing you a strong day!",
    "🧼 Freshly cleaned and ready to go. Let's do this – you're awesome!",
    "🔧 I'm back in shape. Time for your next energy boost.",
    "📎 System's running. This day is bound to go well, right?",
    "😌 Clean, ready – and in a peaceful coffee mood.",
    "☀️ Good morning! I'm ready. Are you?",
    "👋 Back online. I believe in you (and caffeine).",
    "🥸 Ready to brew. Let's do great things – or at least stay awake.",
    "🌞 Ready to brighten your day. Coffee coming right up.",
    "📈 All set. May your day run as smoothly as I do now."
]

The Code

I had a fairly simple setup: One python script to train the model, a folder with my dataset (box / no_box), and a python script to detect and run within a systemd service:

train.py

from tensorflow.keras import layers, models
import numpy as np
import cv2
import os
from sklearn.model_selection import train_test_split

def load_images_from_folder(folder, label):
    images = []
    labels = []
    for filename in os.listdir(folder):
        img_path = os.path.join(folder, filename)
        img = cv2.imread(img_path, cv2.IMREAD_GRAYSCALE)
        if img is not None:
            img = cv2.resize(img, (128, 128)) # Resizing the image to a uniform size
            images.append(img)
            labels.append(label)
    return images, labels

def load_dataset():
    images_with_box, labels_with_box = load_images_from_folder("dataset/box", 1)
    images_without_box, labels_without_box = load_images_from_folder("dataset/no_box", 0)

    images = np.array(images_with_box + images_without_box)
    labels = np.array(labels_with_box + labels_without_box)

    images = images / 255.0
    images = images.reshape(-1, 128, 128, 1)

    return train_test_split(images, labels, test_size=0.2, random_state=42)

def build_model():
    model = models.Sequential([
        layers.Conv2D(32, (3,3), activation='relu', input_shape=(128, 128, 1)),
        layers.MaxPooling2D((2,2)),
        layers.Conv2D(64, (3,3), activation='relu'),
        layers.MaxPooling2D((2,2)),
        layers.Flatten(),
        layers.Dense(128, activation='relu'),
        layers.Dense(1, activation='sigmoid')
    ])

    model.compile(optimizer='adam',
                  loss='binary_crossentropy',
                  metrics=['accuracy'])
    return model

print("Start")

# Loading test data
X_train, X_test, y_train, y_test = load_dataset()

# Creating the model
model = build_model()

# Training the model
model.fit(X_train, y_train, epochs=10, validation_data=(X_test, y_test))

model.save("box_detector_model.h5")

print("Training completed!")

detect.py

import time
import http.client
import requests
import tensorflow as tf
import numpy as np
import json
import cv2
import os
import random
from datetime import datetime

tf.get_logger().setLevel('INFO')

# A simple hysteresis threshold to stop the script from spamming

DETECTION_TIME_THRESHOLD = 40.0 # Seconds with box, before the status changes
ABSENCE_TIME_THRESHOLD = 30.0 # Seconds without box, before the status changes

SLACK_TOKEN = "xoxb-...."
SLACK_CHANNEL = "#your-channel-name"

SAVE_DIR = "./false_positives"

os.makedirs(SAVE_DIR, exist_ok=True)

def send_slack_message(message):
    conn = http.client.HTTPSConnection("slack.com")

    headers = {
        "Authorization": f"Bearer {SLACK_TOKEN}",
        "Content-Type": "application/json",
    }

    data = json.dumps({
        "channel": SLACK_CHANNEL,
        "text": message,
    })

    conn.request("POST", "/api/chat.postMessage", body=data, headers=headers)

    response = conn.getresponse()
    if response.status != 200:
        print(f"Error sending the slack message: {response.read().decode()}")
    else:
        response_data = json.loads(response.read().decode())
        if not response_data.get("ok"):
            print(f"Error sending the slack message: {response_data}")

    conn.close()

def capture_and_predict(model, url):
    response = requests.get(url, stream=True)
    if response.status_code == 200:
        img_array = np.asarray(bytearray(response.content), dtype=np.uint8)
        img_color = cv2.imdecode(img_array, cv2.IMREAD_COLOR)
        img_gray = cv2.cvtColor(img_color, cv2.COLOR_BGR2GRAY)
        img_resized = cv2.resize(img_gray, (128, 128)) / 255.0
        img_input = np.expand_dims(img_resized, axis=-1)
        img_input = np.expand_dims(img_input, axis=0)
        prediction = model.predict(img_input, verbose=0)[0][0]
        return prediction > 0.5, img_color
    else:
        print("Error fetching the image.")
        return None, None

_last_message = None

def get_coffee_message(is_positive=True):
    global _last_message

    messages = [
        "🧼 I'm at the spa. It's called luxury. For you: cleaning mode.",
        "😶‍🌫️ Too much of your love is stuck on me. Let me scrub that off.",
        "🛁 Spa day for me. You guys take a break.",
        "😐 No, I'm not broken. I'm busy with a bubble bath.",
        "🧽 I'm cleaning myself. Why? Because you treat me like an animal.",
        "💤 Cleaning in progress. Please don't knock, I'm sensitive.",
        "🙃 Hygiene is important. That's what my therapist told me.",
        "🚿 I'm showering. You should try it too.",
        "🤖 Maintenance ongoing. Mood? Somewhere between resigned and soapy.",
        "🫣 I'm getting rid of your coffee sins. Don't look."
    ] if is_positive else [
        "☕ Ready for the next coffee. Wishing you a strong day!",
        "🧼 Freshly cleaned and ready to go. Let's do this – you're awesome!",
        "🔧 I'm back in shape. Time for your next energy boost.",
        "📎 System's running. This day is bound to go well, right?",
        "😌 Clean, ready – and in a peaceful coffee mood.",
        "☀️ Good morning! I'm ready. Are you?",
        "👋 Back online. I believe in you (and caffeine).",
        "🥸 Ready to brew. Let's do great things – or at least stay awake.",
        "🌞 Ready to brighten your day. Coffee coming right up.",
        "📈 All set. May your day run as smoothly as I do now."
    ]

    message = random.choice(messages)

    while message == _last_message:
        message = random.choice(messages)

    _last_message = message
    return message

# Loading the model
model = tf.keras.models.load_model("box_detector_model.h5")

box_detected = False
last_detection_time = None
last_absence_time = None

previous_state = None

live_image_url = "http://<the raspi ip>/current"

while True:
    detected, last_image = capture_and_predict(model, live_image_url)
    current_time = time.time()

    if detected != previous_state:
        print('box detected' if detected else 'no box')
        previous_state = detected

    if detected is None:
        time.sleep(1)
        continue

    if detected:
        last_absence_time = None  # reset the opposite tracker

        if not box_detected:
            if last_detection_time is None:
                last_detection_time = current_time
            elif current_time - last_detection_time >= DETECTION_TIME_THRESHOLD:
                box_detected = True
                send_slack_message(get_coffee_message(True))

                # 💾 Bild lokal speichern
                if last_image is not None:
                    timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
                    filename = f"{SAVE_DIR}/box_{timestamp}.jpg"
                    cv2.imwrite(filename, last_image)
                    print(f"Image saved: {filename}")
        else:
            # Box already detected, do nothing special
            pass

    else:
        last_detection_time = None  # reset the opposite tracker

        if box_detected:
            if last_absence_time is None:
                last_absence_time = current_time
            elif current_time - last_absence_time >= ABSENCE_TIME_THRESHOLD:
                box_detected = False
                send_slack_message(get_coffee_message(False))
        else:
            # Box already not detected, do nothing special
            pass

    time.sleep(1)

The Result: A Smarter Office (and Happier Humans)

What started as a spontaneous joke turned into a full-blown IoT project with real impact on daily happiness — and a sprinkle of machine learning magic.

The CoffeeCam is a reminder that tech doesn’t have to be massive or complex to be useful. Sometimes, all it takes is a Pi, a webcam, and a good dose of curiosity.

And now, I’ll get myself a cup of coffee. Cheers!

Automatically Generate Bruno API Collections from Your Symfony Routes

Tim Nelles — Fri, 21 Feb 2025 18:39:31 +0000

Introduction

Bruno is a great and open-source API client designed for efficient testing and collaboration. However, setting up API collections manually can be tedious, especially when dealing with large Symfony applications. Wouldn’t it be great if you could generate Bruno collections automatically from your Symfony routes? This is exactly what opctim/symfony-bruno-generator does.

In this article, we’ll explore how this Symfony bundle can streamline API collection generation, ensuring consistency across projects while saving valuable development time.

Features

opctim/symfony-bruno-generator is a Symfony bundle built using opctim/bruno-lang, a framework-agnostic bru-lang implementation for handling Bruno collections programmatically. This bundle automatically scans your controllers, extracts route information, and generates a Bruno collection in a standardized format.

Automatic Route Extraction: Detects and processes all defined Symfony routes.
Minimal User Input: Guides you through generating API collections interactively.
Customizable Base URL: Allows flexible API testing environments.

Installation

To start using symfony-bruno-generator, install the package via Composer:

composer require --dev opctim/symfony-bruno-generator

Then, register the bundle in config/bundles.php (if you’re not using symfony/flex):

return [
    Opctim\BrunoGeneratorBundle\OpctimBrunoGeneratorBundle::class => ['dev' => true],
];

Usage

Generating a Bruno collection is as simple as running the following command:

php bin/console make:bruno

This command:

Scans all controllers in your Symfony project (below the App\ namespace)
Extracts available routes and methods (GET, POST, etc.)
Writes request files into a structured Bruno collection (retains the controller folder structure in bruno)

Interactive CLI Workflow

During execution, the command will:

Check for an existing Bruno collection
Offer to create a new collection if none exists
Ask for a collection name and base URL
List detected controllers
Prompt you to confirm which requests should be generated

Here’s an example of the output:

Found bruno collection "my_collection" at /bruno

Do you want to generate 5 requests for the UserController? (yes/no) [yes]:

✔ Generated


GET {{baseUrl}}/api/users -> bruno/user/get_api_users.bru
POST {{baseUrl}}/api/users -> bruno/user/post_api_users.bru

Future Improvements

While the current version efficiently extracts routes and generates basic requests, future enhancements include:

Automatic detection of request body structures.
Extraction of query parameters for GET requests.
Support for API Platform metadata.
Custom header configurations for authentication.

Conclusion

With opctim/symfony-bruno-generator, you no longer need to manually maintain your API collection. This package allows Symfony developers to generate Bruno requests effortlessly, ensuring consistency and reducing setup time.

If you’re interested in making your API development workflow smoother, give my package a try. You can also explore opctim/bruno-lang if you need deeper control over Bruno collections.

https://github.com/opctim/symfony-bruno-generator

https://github.com/opctim/bruno-lang

🚀 Happy generating!

Using Symfony’s HeaderBag as a Service: A Debugging Superpower in API Contexts

Tim Nelles — Thu, 21 Nov 2024 12:26:50 +0000

Introduction

Building APIs means dealing with complex layers of cache, middleware, and performance optimizations. While these layers help in production, they can also obscure the actual flow of data. How do you know whether a cache has been hit or missed in a particular request? One simple way is to add custom headers to the response, giving you valuable insights.

In this article, I’ll show you how Symfony’s HeaderBag, when registered as a service, can be a powerful tool for debugging APIs in production. By attaching custom headers that indicate cache hits and misses, you can instantly understand what’s happening in your API without adding unnecessary logging or breaking the flow of execution.

Why HeaderBag?

Symfony’s HeaderBag is part of the HttpFoundation component and is used internally by Symfony’s request and response objects. It provides a structured way to interact with HTTP headers, allowing you to add, modify, and retrieve headers in a clean, object-oriented manner. This makes it the perfect fit for managing custom headers during API execution, especially in debugging scenarios.

But there’s more: you can register HeaderBag as a service and use it across your application, keeping your code clean and your debugging consistent.

Step 1: Registering HeaderBag as a Service

You can easily register HeaderBag as a service in one of two ways.

Option 1: Extend HeaderBag in a Custom Service

You can create a custom service that extends HeaderBag and then use it in your application:

<?php
declare(strict_types=1);

namespace App\Service;

use Symfony\Component\HttpFoundation\HeaderBag;

class ResponseHeaderService extends HeaderBag
{
}

Option 2: Define it in services.yaml

Alternatively, you can directly register HeaderBag as a service in services.yaml:

my.awesome.headerbag:
    class: 'Symfony\Component\HttpFoundation\HeaderBag'

Both methods achieve the same goal: Allowing HeaderBag to be available as a service throughout your app.

Step 2: Adding Custom Headers via an Event Subscriber

Once you’ve registered HeaderBag as a service, the next step is to make it interact with Symfony’s response object. This is where event subscribers come into play. By subscribing to the KernelEvents::RESPONSE event, you can inject custom headers into the response before it’s sent back to the client.

Here’s an example of how to create an event subscriber that adds headers stored in HeaderBag:

<?php
declare(strict_types=1);

namespace App\EventSubscriber;

use App\Service\ResponseHeaderService;
use Symfony\Component\EventDispatcher\EventSubscriberInterface;
use Symfony\Component\HttpKernel\Event\ResponseEvent;
use Symfony\Component\HttpKernel\KernelEvents;

readonly class ResponseHeadersEventSubscriber implements EventSubscriberInterface
{
    public function __construct(
        private ResponseHeaderService $responseHeaderService
    )
    {}

    public static function getSubscribedEvents(): array
    {
        return [
            KernelEvents::RESPONSE => 'onKernelResponse'
        ];
    }

    public function onKernelResponse(ResponseEvent $event): void
    {
        $response = $event->getResponse();

        $response->headers->add($this->responseHeaderService->all());
    }
}

If you preferred to register your HeaderBag as a service in services.yaml, you’ll need to use the Autowire attribute:

<?php
declare(strict_types=1);

namespace App\EventSubscriber;

use Symfony\Component\DependencyInjection\Attribute\Autowire;
use Symfony\Component\EventDispatcher\EventSubscriberInterface;
use Symfony\Component\HttpFoundation\HeaderBag;
use Symfony\Component\HttpKernel\Event\ResponseEvent;
use Symfony\Component\HttpKernel\KernelEvents;

readonly class ResponseHeadersEventSubscriber implements EventSubscriberInterface
{
    public function __construct(
        #[Autowire(service: 'my.awesome.headerbag')]
        private HeaderBag $headerBag
    )
    {}

    // ...

Step 3: Implementing Cache Diagnostics

Now comes the fun part: using this service to track cache hits and misses. Here’s an example of a service that computes some result, checks for cache hits, and adds a custom header to the response indicating whether the cache was hit or missed:

<?php
declare(strict_types=1);

namespace App\Service;

use Psr\Cache\CacheItemPoolInterface;
use Symfony\Component\DependencyInjection\Attribute\Autowire;
use Symfony\Component\HttpFoundation\HeaderBag;

readonly class UniverseService
{
    public function __construct(
        #[Autowire(service: 'my.awesome.headerbag')]
        private HeaderBag $headerBag,
        private CacheItemPoolInterface $cacheItemPool
    )
    {}

    public function calculateUniverseWeight(): float
    {
        $cacheItem = $this->cacheItemPool->getItem('some_cached_data');

        // Check for a cache hit
        if ($cacheItem->isHit() && $cacheItem->get()) {
            // Set the HIT header
            $this->headerBag->set('X-Universe-Cache', 'HIT');

            return $cacheItem->get();
        }

        // Set the MISS header
        $this->headerBag->set('X-Universe-Cache', 'MISS');

        // Perform your heavy lifting...

        return INF;
    }
}

In this example, every time the UniverseService interacts with the cache, it sets an X-Universe-Cache header, indicating whether the cache was hit or missed. This header is then automatically added to the response by the event subscriber. The result? Every API response will contain a simple, informative header that gives you immediate insight into whether the cache was used.

Your Response could look something like this:

HTTP/1.1 200 OK

Date: Mon, 08 Oct 2024 14:23:45 GMT

Server: Apache/2.4.41 (Ubuntu)

Content-Type: application/json

Connection: keep-alive

Content-Length: ∞

X-Universe-Cache: HIT 

{

    "status": "success",

    "message": "The universe's weight has been calculated.",

    "data": {

        "universe_weight": ∞

    }

}

Conclusion

Using Symfony’s HeaderBag as a service is a powerful tool in API development, especially when you’re running in production environments with complex caching layers. By adding custom headers to your API responses, you can quickly determine cache statuses, improving your debugging and optimization processes.

Next time you find yourself wondering whether a particular request was cached, take a look at the headers you’ve added — you’ll have your answer right there!

Syncing an NPM Package Between Multiple Registries

Tim Nelles — Thu, 21 Nov 2024 12:23:11 +0000

Photo by Nicolas Radzimski on Unsplash

The Problem

Our team relies on an internal GitLab NPM registry for managing packages. This works perfectly for internal distribution, but there are instances where a package needs to be shared with external collaborators. Syncing the package from the internal registry to an external one became a new challenge.

I set out to find a solution and, as usual, encountered a series of hurdles that turned this into another chapter in my developer struggles.

Attempt 1: Using HTTP Requests to Mimic NPM CLI

Initially, I thought I could handle this with raw HTTP requests, essentially mimicking what the NPM CLI does under the hood. The idea was simple: query the source registry for package metadata, fetch the tarball, and upload it to the target registry.

Why it failed:

• The process was too complex and fragile.
• It required manually constructing the entire package.json manifest for every version.
• Every small misstep led to broken packages.

Attempt 2: Leveraging npm-registry-sync

Next, I discovered npm-registry-sync, a library designed for syncing NPM registries. This tool almost solved the problem; it could monitor changes and replicate them across registries.

Why it didn’t work for me:

• It operates in “daemon mode,” polling for updates continuously.
• In a GitLab CI pipeline, I needed a one-off execution, controlled entirely by the pipeline — no background processes allowed.

The Winning Solution: The NPM CLI

Eventually, I realized I could stick to the tried-and-true NPM CLI. The steps were straightforward:

Install the package locally from Registry A.
1. Reconfigure NPM to point to Registry B.
2. Publish the package to Registry B.

While this worked like a charm, there were a few extra steps needed to make it CI-friendly.

Configuring NPM for Multiple Registries

Managing registry configurations dynamically in a CI pipeline was a bit tricky.

Here’s how I solved it:

General Configuration

Using the npm CLI, you can set parameters for each registry:

npm config set "//my.awesome.registry.com:<parameter name>=<parameter value>"

Important Gotcha:
The URL in the config must exclude the protocol (https:).

Package-to-Registry Association

To associate a specific namespace or package with a registry:

npm config set "<your namespace>:registry" "<your registry url with https:>"

Handling Authentication in CI Pipelines

Some registries required a username/password combo, while others used tokens. Here’s what I learned:

Tokens

Tokens are straightforward, but ensure you strip the protocol when configuring the auth URL:

npm config set "//my.registry.com:_authToken=<token>"

Basic Auth

Generating a basic auth hash (username:password) required attention to detail. In some distros, the base64 command has quirks that differ from others.

On macOS:

echo -n "<my username & password hash>" | base64

Will give you (as expected):

d2hhdCBhcmUgeW91IGRvaW5nIGhlcmU/IGdvdCB5YSEgc29tZSBtb3JlIHRleHQgdG8gbWFrZSB0aGlzIHJlYWxseSByZWFsbHkgbG9uZw==

With docker linuxkit:

d2hhdCBhcmUgeW91IGRvaW5nIGhlcmU/IGdvdCB5YSEgc29tZSBtb3JlIHRleHQgdG8gbWFrZSB0
aGlzIHJlYWxseSByZWFsbHkgbG9uZw==

What is going on here? There is a line break!

On some distros, the wrap parameter has a default set to 76 chars for formatting private keys etc.

It works like this:

echo -n "<my username & password hash>" | base64 --wrap 0

The echo -n is very important. If omitted, echo will add a linebreak to the end of the string and this will manipulate your hash.

The finished script

Putting it all together, the script could look like this:

#!/usr/bin/env bash

# Input validation
if [ "$#" -ne 5 ]; then
  echo "Usage: $0 <source-registry> <target-registry> <source-registry-token> <target-registry-username> <target-registry-password>"
  exit 1
fi

SOURCE_REGISTRY=$1
TARGET_REGISTRY=$2
SOURCE_REGISTRY_TOKEN=$3
TARGET_REGISTRY_USERNAME=$4
TARGET_REGISTRY_PASSWORD=$5

echo "SOURCE_REGISTRY=$SOURCE_REGISTRY"
echo "TARGET_REGISTRY=$TARGET_REGISTRY"
echo "SOURCE_REGISTRY_TOKEN=${SOURCE_REGISTRY_TOKEN:0:10}"
echo "TARGET_REGISTRY_USERNAME=${TARGET_REGISTRY_USERNAME}"

echo "Syncing from '$SOURCE_REGISTRY' to '$TARGET_REGISTRY'..."

remove_protocol() {
  sed -E 's/^https?://g;t' <<< "$1"
}

packages=(
  "@my-namespace/my-package"
)

# INFO: Removing the protocol (^https:) from the authToken config is crucial. Will not work without it.

# Auth for source registry
npm config set "$(remove_protocol $SOURCE_REGISTRY):_authToken=$SOURCE_REGISTRY_TOKEN"

# --wrap 0 is super important! If omitted, it will cause the base64 to contain a line break after 76 chars!
BASIC_AUTH=$(echo -n "$TARGET_REGISTRY_USERNAME:$TARGET_REGISTRY_PASSWORD" | base64 --wrap 0)

# Auth for target registry
npm config set "$(remove_protocol $TARGET_REGISTRY):_auth" "$BASIC_AUTH"
npm config set "$(remove_protocol $TARGET_REGISTRY):always-auth=true"

# Set to source registry to fetch metadata
npm config set "@my-namespace:registry" "$SOURCE_REGISTRY"

for package in "${packages[@]}"; do
  echo "Syncing '$package'..."

  for version in $(npm view "$package" --json | jq -r '.versions[]'); do
    # Install locally
    npm install "$package@${version}" --ignore-scripts

    # Switch to target registry
    npm config set "@my-namespace:registry" "$TARGET_REGISTRY"

    # Publish on target
    npm publish ./node_modules/"$package"

    # Re-set to source registry
    npm config set "@my-namespace:registry" "$SOURCE_REGISTRY"
  done
done

echo "Done syncing packages."

So you could use it like this:

#!/usr/bin/env bash

SOURCE_REGISTRY=https://my.source.registry.com
SOURCE_REGISTRY_TOKEN=abc-xyz-1234568
TARGET_REGISTRY=https://my.target.registry.com
TARGET_REGISTRY_USERNAME=john_doe
TARGET_REGISTRY_PASSWORD=supersecret123

sync-registries.sh $SOURCE_REGISTRY $TARGET_REGISTRY $SOURCE_REGISTRY_TOKEN $TARGET_REGISTRY_USERNAME $TARGET_REGISTRY_PASSWORD

Key Lessons Learned

Stick to Simple Tools:
The npm CLI might not be fancy for this task, but it’s reliable and gets the job done.
Mind the Details:
Configuring authentication, especially with base64, can have subtle platform-specific quirks.
Keep It CI-Friendly:
Avoid solutions like daemons or background tasks when working in CI/CD pipelines. Keep the process under pipeline control.

Syncing npm packages between registries was a frustrating but rewarding learning experience. If you’re facing a similar challenge, I hope these lessons help you navigate the struggle with a bit more ease!