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.
Before using this API you will need to contact Kioubit to have your domain enabled. Use the
'localhost'
domain for testing purposes
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.