The Kioubit Verify Services allow you to verify that a user is the holder of an ASN based on the
authentication
methods the user has provided to the dn42 registry.
Alternatively use OAuth/OIDC to connect your application to the Kioubit Authentication Services: Control Panel
1. Add the Kioubit Authentication Button to your website
Replace https://example.org/ with your website where the
code
from step 2 is
hosted It is also possible to include the optional user-defined 'token' parameter. Download the full required
styles,
including the examples.
2. Add the following verification code to your website
<?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();
1) Add this HTML page to your website:
<!DOCTYPE html>
<html>
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Authenticate with Kioubit.dn42</title>
<script src="https://telegram.org/js/telegram-web-app.js"></script>
<script src="kioubit-auth-button.js"></script>
<style>
body { font-family: sans-serif; text-align: center; padding-top: 50px; }
.loading { display: none; color: #666; }
kioubit-auth-btn { display: block; width: fit-content; margin: 0 auto; }
</style>
</head>
<body>
<div id="ui">
<kioubit-auth-btn return="" token=""></kioubit-auth-btn>
</div>
<div id="loader" class="loading">
<p>Sending data back to bot...</p>
</div>
<script>
const tg = window.Telegram.WebApp;
tg.ready();
const urlParams = new URLSearchParams(window.location.search);
if (urlParams.has('params')) {
document.getElementById('ui').style.display = 'none';
document.getElementById('loader').style.display = 'block';
// Prepare data object
const result = {};
urlParams.forEach((value, key) => result[key] = value);
// Send data back to the telegram bot
// This closes the Mini App and triggers the 'web_app_data' event in the bot
tg.sendData(JSON.stringify(result));
}
</script>
</body>
</html>
2) Add the following logic to your bot:
const { Telegraf, Markup } = require('telegraf');
// 1. Paste your token from @BotFather here
const bot = new Telegraf('');
// 2. Paste the URL where you hosted your index
const APP_URL = 'https://example.org/temp/telegram.html';
bot.command('verify', (ctx) => {
console.log(ctx);
ctx.reply('Open the app to verify:',
Markup.keyboard([
Markup.button.webApp('🚀 Start Verification', APP_URL)
]).resize()
);
});
bot.on('web_app_data', (ctx) => {
try {
// Get the raw string from the Telegram message
const rawJsonString = ctx.message.web_app_data.data;
// Parse the outer JSON (contains "params" and "signature")
const envelope = JSON.parse(rawJsonString);
// Decode the "params" field from Base64 to a UTF-8 string
const decodedString = Buffer.from(envelope.params, 'base64').toString('utf-8');
// Parse the actual Kioubit data
const data = JSON.parse(decodedString);
// Important: Verify the data here, using the Node.JS method
// Send the message to the chat
const report = `
✅ **ASN Verification Received**
• **ASN:** ${data.asn}
• **Maintainer:** ${data.effective_mnt}
• **Method:** ${data.authtype}
• **Domain:** ${data.domain}
`;
console.log(report);
ctx.replyWithMarkdown(report);
} catch (e) {
console.error("Failed to process data:", e);
ctx.reply("Error: Data was received but could not be decoded. Check console.");
}
});
bot.launch();
// Enable graceful stop
process.once('SIGINT', () => bot.stop('SIGINT'));
process.once('SIGTERM', () => bot.stop('SIGTERM'));
console.log('Bot is running...');
3. Download Kioubit's Public Key required for the verification code
Download the public key and place
it
in the
same directory as the verification code.