Kioubit Authentication Services

<?php

// Get parameters and signature
$params = filter_var($_GET["params"], FILTER_SANITIZE_URL);
$signature = filter_var($_GET["signature"], FILTER_SANITIZE_URL);

// Signature Verification
$public_key_pem = file_get_contents('public_key.pem');
$r = openssl_verify($params, base64_decode($signature), $public_key_pem, OPENSSL_ALGO_SHA512);
if ($r != 1) {
    die("Could not verify signature");
}

// Decode parameters to json
$params = base64_decode($params);
$info = json_decode($params, true);

// Check that the request is for our own domain
if ("owndomain.com" != $info["domain"]) {
    die("The request is for a different domain");
}

// Check the time of the request
if (abs($info["time"] - time()) > 60) {
    die("The request has expired");
}

// Print json object
echo nl2br(print_r($info, true));

?>

#!/usr/bin/env python3

import base64
import os
import json
import time
# Requires pyca/cryptography (cryptography==43.0.1)
from cryptography import x509
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.exceptions import InvalidSignature


class AuthVerifier():
    PUBKEY_FILE = os.path.dirname(__file__)+"/public_key.pem"
    __domain: str = ""
    __pubKey: ec.EllipticCurvePublicKey | None = None

    def __init__(self, domain: str, pubkey: str = PUBKEY_FILE):
        self.__domain = domain
        with open(pubkey, "br") as pk:
            pk_content = pk.read()
        self.__pubKey = serialization.load_pem_public_key(pk_content)

    def verify(self, params: str, signature: str) -> tuple[bool, dict[str, any] | str]:
        try:
            sig = base64.b64decode(signature)
        except base64.binascii.Error:
            return False, "Invalid base64 encoding for 'signature'"
        try:
            self.__pubKey.verify(sig, params.encode(), ec.ECDSA(hashes.SHA512()))
        except InvalidSignature:
            return False, "Invalid signature"

        try:
            params_decoded = base64.b64decode(params)
        except base64.binascii.Error:
            return False, "Invalid base64 encoding for 'params'"

        try:
            user_data = json.loads(params_decoded)
            if abs(time.time() - user_data["time"]) > 60:
                return False, "Signature has expired"
            if user_data["domain"] != self.__domain.replace("https://",""):
                return False, "Invalid domain"
        except json.decoder.JSONDecodeError:
            return False, "Invalid user data JSON"
        except KeyError:
            return False, "Required value not found in JSON"
        return True, user_data


if __name__ == "__main__":
    test_params="eyJhc24iOiI0MjQyNDIzMDM1IiwidGltZSI6MTY2ODI2NjkyNiwiYWxsb3dlZDQiOiIxNzIuMjIuMTI1LjEyOFwvMjYsMTcyLjIwLjAuODFcLzMyIiwiYWxsb3dlZDYiOiJmZDYzOjVkNDA6NDdlNTo6XC80OCxmZDQyOmQ0MjpkNDI6ODE6OlwvNjQiLCJtbnQiOiJMQVJFLU1OVCIsImF1dGh0eXBlIjoibG9naW5jb2RlIiwiZG9tYWluIjoic3ZjLmJ1cmJsZS5kbjQyIn0="
    test_signature="MIGIAkIBAmwz3sQ1vOkH8+8e0NJ8GsUqKSaazIWmYDp60sshlTo7gCAopZOZ6/+tD6s+oEGM1i5mKGbHgK9ROATQLHxUZecCQgCa2N828uNn76z1Yg63/c7veMVIiK4l1X9TCUepJnJ3mCto+7ogCP+2vQm6GHipSNRF4wnt6tZbir0HZvrqEnRAmA=="
    
    from unittest import mock
    example_verifier = AuthVerifier("svc.burble.dn42")

    @mock.patch('time.time', mock.MagicMock(return_value=1668266926))
    def test_with_correct_time():
        result, user_data = example_verifier.verify(params=test_params, signature=test_signature)
        print(user_data)
        assert result == True
        
    @mock.patch('time.time', mock.MagicMock(return_value=1668266910))
    def test_with_backwards_time():
        result, _ = example_verifier.verify(params=test_params, signature=test_signature)
        assert result == True

    @mock.patch('time.time', mock.MagicMock(return_value=1668266000))
    def test_with_expired_time():
        result, _ = example_verifier.verify(params=test_params, signature=test_signature)
        assert result == False

    test_with_correct_time()
    test_with_backwards_time()
    test_with_expired_time()

package main

import (
	"crypto/ecdsa"
	"crypto/sha512"
	"crypto/x509"
	_ "embed"
	"encoding/base64"
	"encoding/json"
	"encoding/pem"
	"errors"
	"fmt"
	"math"
	"time"
)

//go:embed public_key.pem
var pemPubKey []byte

const myDomain = "svc.burble.dn42"

func VerifyAuthToken(signature, params string) (user_data map[string]any, err error) {
	// Read public key
	blockPub, _ := pem.Decode(pemPubKey)
	if blockPub == nil {
		return nil, errors.New("failed to decode PEM block")
	}
	x509EncodedPub := blockPub.Bytes
	genericPublicKey, err := x509.ParsePKIXPublicKey(x509EncodedPub)
	if err != nil {
		return nil, errors.New("internal server error")
	}
	publicKey := genericPublicKey.(*ecdsa.PublicKey)

	// Hash parameters
	hash := sha512.Sum512([]byte(params))

	// Decode base64 signature
	signatureBytes, err := base64.StdEncoding.DecodeString(signature)
	if err != nil {
		return nil, errors.New("failed to decode signature")
	}

	// Verify signature
	if !ecdsa.VerifyASN1(publicKey, hash[:], signatureBytes) {
		return nil, errors.New("invalid signature")
	}

	// Decode parameters
	parameterBytes, err := base64.StdEncoding.DecodeString(params)
	if err != nil {
		return nil, fmt.Errorf("failed decoding verified parameters: %w", err)
	}

	err = json.Unmarshal(parameterBytes, &user_data)
	if err != nil {
		return nil, fmt.Errorf("failed unmarshaling verified parameters: %w", err)
	}

	if math.Abs(user_data["time"].(float64)-float64(time.Now().Unix())) > 60 {
		return nil, errors.New("the request has expired")
	}

	if user_data["domain"].(string) != myDomain {
		return nil, errors.New("this request is for a different domain")
	}

	return
}

func main() {
	var test_params = "eyJhc24iOiI0MjQyNDIzMDM1IiwidGltZSI6MTY2ODI2NjkyNiwiYWxsb3dlZDQiOiIxNzIuMjIuMTI1LjEyOFwvMjYsMTcyLjIwLjAuODFcLzMyIiwiYWxsb3dlZDYiOiJmZDYzOjVkNDA6NDdlNTo6XC80OCxmZDQyOmQ0MjpkNDI6ODE6OlwvNjQiLCJtbnQiOiJMQVJFLU1OVCIsImF1dGh0eXBlIjoibG9naW5jb2RlIiwiZG9tYWluIjoic3ZjLmJ1cmJsZS5kbjQyIn0="
	var test_signature = "MIGIAkIBAmwz3sQ1vOkH8+8e0NJ8GsUqKSaazIWmYDp60sshlTo7gCAopZOZ6/+tD6s+oEGM1i5mKGbHgK9ROATQLHxUZecCQgCa2N828uNn76z1Yg63/c7veMVIiK4l1X9TCUepJnJ3mCto+7ogCP+2vQm6GHipSNRF4wnt6tZbir0HZvrqEnRAmA=="
	fmt.Println(VerifyAuthToken(test_signature, test_params))
}

import { readFileSync } from 'fs';
import { createPublicKey, createVerify } from 'crypto';

class KioubitAuthVerifier {
  static #PUBKEY_FILE = "public_key.pem";
  #pubKey;
  #domain;

  constructor(domain, pubkey = this.constructor.#PUBKEY_FILE) {
    this.#domain = domain;

    const pubKeyContent = readFileSync(pubkey);
    this.#pubKey = createPublicKey(pubKeyContent);
  }

  verify(params, signature) {
    let sig;
    try {
      sig = Buffer.from(signature, 'base64');
    } catch (e) {
      return [false, "Invalid base64 encoding for 'signature'"];
    }

    // Verify the signature
    try {
      const sigVerify = createVerify('SHA512');
      sigVerify.update(Buffer.from(params, 'utf8'));
      sigVerify.end();

      const isValid = sigVerify.verify(this.#pubKey, sig);
      if (!isValid) {
        return [false, "Invalid signature"];
      }
    } catch (e) {
      return [false, "Error verifying the signature"];
    }

    // Decode the params
    let user_data;
    try {
      const paramsDecoded = Buffer.from(params, 'base64');
      user_data = JSON.parse(paramsDecoded.toString('utf8'));
    } catch (e) {
      return [false, "Invalid base64 encoding for 'params'"];
    }

    // Check user data validity
    if (Math.abs(Date.now() / 1000 - user_data.time) > 60) {
      return [false, "Signature has expired"];
    }

    if (user_data.domain !== this.#domain) {
      return [false, "Request is for a different domain"];
    }

    return [true, user_data];
  }
}

const testParams = "eyJhc24iOiI0MjQyNDIzMDM1IiwidGltZSI6MTY2ODI2NjkyNiwiYWxsb3dlZDQiOiIxNzIuMjIuMTI1LjEyOFwvMjYsMTcyLjIwLjAuODFcLzMyIiwiYWxsb3dlZDYiOiJmZDYzOjVkNDA6NDdlNTo6XC80OCxmZDQyOmQ0MjpkNDI6ODE6OlwvNjQiLCJtbnQiOiJMQVJFLU1OVCIsImF1dGh0eXBlIjoibG9naW5jb2RlIiwiZG9tYWluIjoic3ZjLmJ1cmJsZS5kbjQyIn0=";
const testSignature = "MIGIAkIBAmwz3sQ1vOkH8+8e0NJ8GsUqKSaazIWmYDp60sshlTo7gCAopZOZ6/+tD6s+oEGM1i5mKGbHgK9ROATQLHxUZecCQgCa2N828uNn76z1Yg63/c7veMVIiK4l1X9TCUepJnJ3mCto+7ogCP+2vQm6GHipSNRF4wnt6tZbir0HZvrqEnRAmA==";
const exampleVerifier = new KioubitAuthVerifier("svc.burble.dn42");

function testWithCorrectTime() {
  // Mock current time for testing
  const originalDateNow = Date.now;
  Date.now = () => 1668266926 * 1000;

  const [result, userData] = exampleVerifier.verify(
    testParams,
    testSignature
  );
  console.assert(result === true, "Should be valid with correct time");
  console.log(userData);

  // Restore original Date.now
  Date.now = originalDateNow;
}

testWithCorrectTime();