feat: Initial code

.gitea/workflows/build.yml

@@ -0,0 +1,41 @@
+name: Build Docker
+run-name: Build Docker Images
+on:
+  push:
+
+jobs:
+  Build Image:
+    runs-on: [ubuntu-latest, amd]
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v2
+      - name: Install Docker
+        run : |
+          apt-get install ca-certificates curl gnupg
+          install -m 0755 -d /etc/apt/keyrings
+          curl -fsSL https://download.docker.com/linux/debian/gpg | gpg --dearmor -o /etc/apt/keyrings/docker.gpg
+          chmod a+r /etc/apt/keyrings/docker.gpg
+          echo "deb [arch="$(dpkg --print-architecture)" signed-by=/etc/apt/keyrings/docker.gpg] https://download.docker.com/linux/debian "$(. /etc/os-release && echo "$VERSION_CODENAME")" stable" | tee /etc/apt/sources.list.d/docker.list > /dev/null
+          apt-get update
+          apt-get install docker-ce-cli -y          
+      - name: Build Docker image
+        run : |          
+          echo "${{ secrets.DOCKERGIT_TOKEN }}" | docker login git.woodburn.au -u nathanwoodburn --password-stdin
+          echo "branch=${GITHUB_HEAD_REF:-${GITHUB_REF#refs/heads/}}"
+          tag=${GITHUB_HEAD_REF:-${GITHUB_REF#refs/heads/}}
+          tag=${tag//\//-}
+          tag_num=${GITHUB_RUN_NUMBER}
+          echo "tag_num=$tag_num"
+
+          if [[ "$tag" == "main" ]]; then
+            tag="latest"
+          else
+            tag_num="${tag}-${tag_num}"
+          fi
+          
+
+          docker build -t hip02-resolver:$tag_num .
+          docker tag hip02-resolver:$tag_num git.woodburn.au/nathanwoodburn/hip02-resolver:$tag_num
+          docker push git.woodburn.au/nathanwoodburn/hip02-resolver:$tag_num
+          docker tag hip02-resolver:$tag_num git.woodburn.au/nathanwoodburn/hip02-resolver:$tag
+          docker push git.woodburn.au/nathanwoodburn/hip02-resolver:$tag

.gitignore

@@ -0,0 +1,2 @@
+
+__pycache__/

Dockerfile

@@ -0,0 +1,14 @@
+FROM --platform=$BUILDPLATFORM python:3.10-alpine AS builder
+
+WORKDIR /app
+
+COPY requirements.txt /app
+RUN --mount=type=cache,target=/root/.cache/pip \
+    pip3 install -r requirements.txt
+
+COPY . /app
+
+ENTRYPOINT ["python3"]
+CMD ["server.py"]
+
+FROM builder as dev-envs

hip02.py

@@ -0,0 +1,124 @@
+# DNS
+import binascii
+import datetime
+import subprocess
+import tempfile
+import dns.resolver
+from cryptography import x509
+from cryptography.hazmat.backends import default_backend
+
+
+def resolve(HSD_IP, HSD_PORT, domain, token="HNS"):
+    resolver = dns.resolver.Resolver()
+    resolver.nameservers = [HSD_IP]
+    resolver.port = HSD_PORT
+    records = []
+    try:
+        # Query the DNS record
+        response = resolver.resolve(domain, "A")
+        for record in response:
+            records.append(str(record))
+        if not records:
+            return False
+    except Exception as e:
+        return False
+
+    Server_IP = records[0]
+    curl_command = ["curl","--connect-to",f"{domain}:443:{Server_IP}:443",f"https://{domain}/.well-known/wallets/{token}","--insecure"]
+    curl_process = subprocess.Popen(curl_command, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    curl_output, _ = curl_process.communicate()
+    return curl_output.decode("utf-8")
+
+def TLSA_check(HSD_IP,HSD_PORT,domain):
+    resolver = dns.resolver.Resolver()
+    resolver.nameservers = [HSD_IP]
+    resolver.port = HSD_PORT
+    try:
+        # Query the DNS record
+        response = resolver.resolve(domain, "A")
+        records = []
+        for record in response:
+            records.append(str(record))
+
+        if not records:
+            return "No A record found"
+        
+        # Get the first A record
+        ip = records[0]
+        
+        # Run the openssl s_client command
+        s_client_command = ["openssl","s_client","-showcerts","-connect",f"{ip}:443","-servername",domain,]
+        s_client_process = subprocess.Popen(s_client_command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, stdin=subprocess.PIPE)
+        s_client_output, _ = s_client_process.communicate(input=b"\n")        
+        certificates = []
+        current_cert = ""
+        for line in s_client_output.split(b"\n"):
+            current_cert += line.decode("utf-8") + "\n"
+            if "-----END CERTIFICATE-----" in line.decode("utf-8"):
+                certificates.append(current_cert)
+                current_cert = ""
+        # Remove anything before -----BEGIN CERTIFICATE-----
+        certificates = [cert[cert.find("-----BEGIN CERTIFICATE-----"):] for cert in certificates]
+        if not certificates:
+            return "No certificates found"        
+        cert = certificates[0]
+        with tempfile.NamedTemporaryFile(mode="w", delete=False) as temp_cert_file:
+            temp_cert_file.write(cert)
+            temp_cert_file.seek(0)  # Move back to the beginning of the temporary file
+        tlsa_command = ["openssl","x509","-in",temp_cert_file.name,"-pubkey","-noout","|","openssl","pkey","-pubin","-outform","der","|","openssl","dgst","-sha256","-binary",]        
+        tlsa_process = subprocess.Popen(" ".join(tlsa_command), shell=True, stdout=subprocess.PIPE)
+        tlsa_output, _ = tlsa_process.communicate()
+        tlsa_server = "3 1 1 " + binascii.hexlify(tlsa_output).decode("utf-8")
+
+        # Get domains
+        cert_obj = x509.load_pem_x509_certificate(cert.encode("utf-8"), default_backend())
+
+        domains = []
+        for ext in cert_obj.extensions:
+            if ext.oid == x509.ExtensionOID.SUBJECT_ALTERNATIVE_NAME:
+                san_list = ext.value.get_values_for_type(x509.DNSName)
+                domains.extend(san_list)
+        
+        # Extract the common name (CN) from the subject
+        common_name = cert_obj.subject.get_attributes_for_oid(x509.NameOID.COMMON_NAME)
+        if common_name:
+            if common_name[0].value not in domains:
+                domains.append(common_name[0].value)
+
+        if not domains:
+            return "Invalid certificate"
+
+        domain_parts = domain.split(".")        
+        higher_domain = ""
+        for i in range(1,len(domain_parts)):
+            higher_domain = domain_parts[i] + "." + higher_domain
+        
+        higher_domain = higher_domain[:-1]
+
+        if not domain in domains and not "*." + higher_domain in domains:
+            return "Invalid certificate - Missing domain"
+        
+        expiry_date = cert_obj.not_valid_after
+        # Check if expiry date is past
+        if expiry_date < datetime.datetime.now():
+            return "Invalid certificate - Expired"
+
+        try:
+            # Check for TLSA record
+            response = resolver.resolve("_443._tcp."+domain, "TLSA")
+            tlsa_records = []
+            for record in response:
+                tlsa_records.append(str(record))
+
+            if not tlsa_records:
+                return "No TLSA record found"
+            else:
+                if tlsa_server != tlsa_records[0]:
+                    return "Invalid TLSA record"
+        except:
+            return "Exception in TLSA record check"
+        return True
+        
+    # Catch all exceptions
+    except Exception as e:
+        return "Exception: "+str(e)

main.py

@@ -0,0 +1,71 @@
+from flask import Flask, make_response, redirect, request
+import os
+import dotenv
+import requests
+import hip02
+
+
+app = Flask(__name__)
+dotenv.load_dotenv()
+
+HSD_IP = '10.2.1.15'
+HSD_PORT = 5350
+
+if os.getenv('HSD_IP'):
+    HSD_IP = os.getenv('HSD_IP')
+if os.getenv('HSD_PORT'):
+    HSD_PORT = os.getenv('HSD_PORT')
+
+
+@app.route('/')
+def index():
+    return redirect("https://nathan.woodburn.au")
+
+
+# Special routes
+@app.route('/.well-known/wallets/<token>')
+def send_wallet(token):
+    address = requests.get('https://nathan.woodburn.au/.well-known/wallets/'+token).text
+    return make_response(address, 200, {'Content-Type': 'text/plain'})
+
+@app.route('/favicon.ico')
+def favicon():
+    return redirect('https://nathan.woodburn.au/favicon.ico')
+
+@app.route('/<path>.json')
+def jsonlookup(path):
+    TLSA = hip02.TLSA_check(HSD_IP,HSD_PORT,path)
+    if not TLSA:
+        return make_response({"success":False,"error":TLSA}, 200, {'Content-Type': 'application/json'})
+    
+    token = "HNS"
+    if 'token' in request.args:
+        token = request.args['token'].upper()
+
+    hip2 = hip02.resolve(HSD_IP, HSD_PORT, path,token)
+    if not hip2:
+        return make_response({"success":False,"error":hip2}, 200, {'Content-Type': 'application/json'})
+
+    return make_response({"success":True,"address":hip2}, 200, {'Content-Type': 'application/json'})
+
+
+@app.route('/<path>')
+def lookup(path):
+    TLSA = hip02.TLSA_check(HSD_IP,HSD_PORT,path)
+    if not TLSA:
+        return make_response(TLSA, 200, {'Content-Type': 'text/plain'})
+    token = "HNS"
+    if 'token' in request.args:
+        token = request.args['token'].upper()
+    return make_response(hip02.resolve(HSD_IP, HSD_PORT, path,token), 200, {'Content-Type': 'text/plain'})
+
+# 404 catch all
+@app.errorhandler(404)
+def not_found(e):
+    return redirect('/')
+
+
+
+
+if __name__ == '__main__':
+    app.run(debug=False, port=5000, host='0.0.0.0')

requirements.txt

@@ -0,0 +1,6 @@
+flask
+python-dotenv
+gunicorn
+requests
+dnspython
+cryptography

server.py

@@ -0,0 +1,42 @@
+import time
+from flask import Flask
+from main import app
+import main
+from gunicorn.app.base import BaseApplication
+import os
+import dotenv
+import sys
+import json
+
+
+class GunicornApp(BaseApplication):
+    def __init__(self, app, options=None):
+        self.options = options or {}
+        self.application = app
+        super().__init__()
+
+    def load_config(self):
+        for key, value in self.options.items():
+            if key in self.cfg.settings and value is not None:
+                self.cfg.set(key.lower(), value)
+
+    def load(self):
+        return self.application
+
+if __name__ == '__main__':
+    workers = os.getenv('WORKERS')
+    threads = os.getenv('THREADS')
+    if workers is None:
+        workers = 1
+    if threads is None:
+        threads = 2
+    workers = int(workers)
+    threads = int(threads)
+    options = {
+        'bind': '0.0.0.0:5000',
+        'workers': workers,
+        'threads': threads,
+    }
+    gunicorn_app = GunicornApp(app, options)
+    print('Starting server with ' + str(workers) + ' workers and ' + str(threads) + ' threads', flush=True)
+    gunicorn_app.run()