feat: Add onchain TXT record login using FIRE HSD

This reverts commit eaf416316a.

.python-version

@@ -0,0 +1 @@
+3.13

main.py

@@ -0,0 +1,6 @@
+def main():
+    print("Hello from hns-login!")
+
+
+if __name__ == "__main__":
+    main()

migrate_db.py

@@ -1,10 +0,0 @@
-#!/usr/bin/env python3
-
-from website import create_app
-from website.migrations import add_missing_columns_to_oauth2_code
-
-if __name__ == '__main__':
-    app = create_app()
-    print("Running database migration...")
-    add_missing_columns_to_oauth2_code(app)
-    print("Migration completed.")

pyproject.toml

@@ -0,0 +1,22 @@
+[project]
+name = "hns-login"
+version = "0.1.0"
+description = "Add your description here"
+readme = "README.md"
+requires-python = ">=3.13"
+dependencies = [
+    "authlib>=1.6.5",
+    "dnspython>=2.6.1",
+    "eth-account>=0.13.7",
+    "flask>=3.1.2",
+    "flask-sqlalchemy>=3.1.1",
+    "python-dotenv>=1.2.1",
+    "requests>=2.32.3",
+    "requests-doh>=1.0.0",
+    "web3>=7.14.0",
+]
+
+[dependency-groups]
+dev = [
+    "ruff>=0.14.5",
+]

website/__init__.py

@@ -1,17 +0,0 @@
-import os
-from flask import Flask
-
-def create_app():
-    app = Flask(__name__)
-    app.config.from_object(os.environ['APP_SETTINGS'])
-
-    with app.app_context():
-        # Run migrations first, before any database operations
-        from .migrations import add_missing_columns_to_oauth2_code
-        add_missing_columns_to_oauth2_code(app)
-        
-        # Import models after migration but before init_db
-        from .models import db
-        db.create_all()
-    
-    return app

website/migrations.py

@@ -1,91 +0,0 @@
-import os
-import sqlite3
-from sqlalchemy import inspect
-from flask import current_app
-import logging
-
-def add_missing_columns_to_oauth2_code(app):
-    """
-    Check and add missing columns to oauth2_code table
-    """
-    print("Starting database migration check...")
-    with app.app_context():
-        from website.models import db
-        
-        # Get the engine and inspector
-        engine = db.engine
-        inspector = inspect(engine)
-        
-        # Check if oauth2_code table exists
-        if 'oauth2_code' not in inspector.get_table_names():
-            print("oauth2_code table doesn't exist yet, skipping migration")
-            return  # Table doesn't exist yet
-            
-        # Get existing columns
-        columns = [column['name'] for column in inspector.get_columns('oauth2_code')]
-        print(f"Existing columns in oauth2_code: {columns}")
-        
-        # Define columns that should be added if missing
-        missing_columns = {
-            'acr': 'TEXT',
-            'amr': 'TEXT',
-            'code_challenge': 'TEXT',
-            'code_challenge_method': 'TEXT'
-        }
-        
-        # Check which columns need to be added
-        columns_to_add = {col: dtype for col, dtype in missing_columns.items() if col not in columns}
-        
-        if not columns_to_add:
-            print("No columns need to be added, schema is up to date")
-            return  # No columns need to be added
-        
-        print(f"Columns to add: {columns_to_add}")
-        
-        # Connect directly to SQLite to add columns
-        try:
-            # Get database URI from app config
-            db_uri = current_app.config.get('SQLALCHEMY_DATABASE_URI')
-            print(f"Database URI: {db_uri}")
-            
-            # Handle both relative and absolute paths
-            if db_uri.startswith('sqlite:///'):
-                # Relative path
-                if db_uri.startswith('sqlite:////'):
-                    # Absolute path
-                    db_path = db_uri.replace('sqlite:////', '/')
-                else:
-                    # Relative path - may need to be adjusted for Docker
-                    db_path = os.path.join(app.root_path, '..', db_uri.replace('sqlite:///', ''))
-            else:
-                # Memory or other type of database
-                print(f"Unsupported database type: {db_uri}")
-                return
-                
-            print(f"Attempting to connect to database at: {db_path}")
-            
-            # Ensure directory exists
-            db_dir = os.path.dirname(db_path)
-            if not os.path.exists(db_dir):
-                print(f"Database directory doesn't exist: {db_dir}")
-            
-            # Connect to database
-            conn = sqlite3.connect(db_path)
-            cursor = conn.cursor()
-            
-            for column, dtype in columns_to_add.items():
-                try:
-                    sql = f'ALTER TABLE oauth2_code ADD COLUMN {column} {dtype};'
-                    print(f"Executing SQL: {sql}")
-                    cursor.execute(sql)
-                    print(f"Successfully added column '{column}' to oauth2_code table")
-                except sqlite3.OperationalError as e:
-                    print(f"Error adding column '{column}': {str(e)}")
-            
-            conn.commit()
-            conn.close()
-            print("Migration completed successfully")
-        except Exception as e:
-            print(f"Error during migration: {str(e)}")
-            import traceback
-            traceback.print_exc()

website/routes.py

@@ -1,7 +1,7 @@
 import time
 import datetime as dt
 from .varo_auth import flask_login as varo_auth_flask_login
-from flask import Blueprint, request, session, url_for, make_response
+from flask import Blueprint, request, session, url_for
 from flask import render_template, redirect, jsonify, send_from_directory
 from werkzeug.security import gen_salt
 from authlib.integrations.flask_oauth2 import current_token
@@ -18,7 +18,6 @@ from datetime import timedelta
 from eth_account.messages import encode_defunct
 from eth_account import Account
 import json
-import urllib.parse
 
 
 
@@ -57,24 +56,49 @@ def split_by_crlf(s):
     return [v for v in s.splitlines() if v]
 
 def get_idns_records(domain:str) -> list:
-    query = dns.message.make_query(domain, dns.rdatatype.TXT)
-    dns_request = query.to_wire()
-    # Send the DNS query over HTTPS
-    response = requests.post('https://hnsdoh.com/dns-query', data=dns_request, headers={'Content-Type': 'application/dns-message'})
-    # Parse the DNS response
-    dns_response = dns.message.from_wire(response.content)
-    # Loop over TXT records and look for profile 
     idns_records = []
-    for record in dns_response.answer:
+    try:
-        if record.rdtype == dns.rdatatype.TXT:
+        query = dns.message.make_query(domain, dns.rdatatype.TXT)
-            for txt in record:
+        dns_request = query.to_wire()
-                txt_value:str = txt.to_text().strip('"')
+        # Send the DNS query over HTTPS
-                if txt_value.startswith("IDNS1"):
+        response = requests.post('https://au.hnsdoh.com/dns-query', data=dns_request, headers={'Content-Type': 'application/dns-message'})
-                    print(txt_value)
+        # Parse the DNS response
-                    idns = txt_value.removeprefix("IDNS1 ")
+        dns_response = dns.message.from_wire(response.content)
-                    idns = idns.split(" ")
+        # Loop over TXT records and look for profile 
-                    for r in idns:
+        for record in dns_response.answer:
-                        idns_records.append(r)
+            if record.rdtype == dns.rdatatype.TXT:
+                for txt in record:
+                    txt_value:str = txt.to_text().strip('"')
+                    if txt_value.startswith("IDNS1"):
+                        print(txt_value)
+                        idns = txt_value.removeprefix("IDNS1 ")
+                        idns = idns.split(" ")
+                        for r in idns:
+                            idns_records.append(r)
+    except Exception as e:
+        print(f"Error fetching DNS records: {e}")
+
+    # Get onchain records
+    try:
+        onchain_response = requests.get(f"https://hsd.hns.au/api/v1/nameresource/{domain}")
+        if onchain_response.status_code == 200:
+            onchain_data = onchain_response.json()
+            if "records" in onchain_data:
+                for record in onchain_data["records"]:
+                    if record["type"] == "TXT":
+                        txt_values = record["txt"]
+                        for txt_value in txt_values:
+                            txt_value = txt_value.strip('"')
+                            if txt_value.startswith("IDNS1"):
+                                print(txt_value)
+                                idns = txt_value.removeprefix("IDNS1 ")
+                                idns = idns.split(" ")
+                                for r in idns:
+                                    idns_records.append(r)
+        
+    except Exception as e:
+        print(f"Error fetching onchain records: {e}")
+
     return idns_records
 
 def get_user_info(user:User) -> dict:
@@ -145,7 +169,7 @@ def home():
 
     if request.method == "POST":
         auth = varo_auth_flask_login(request)
-        if auth == False:
+        if not auth:
             return redirect("/?error=login_failed")
         print(auth)
         user = User.query.filter_by(username=auth).first()
@@ -267,32 +291,55 @@ def hnsid_domain(domain):
 
 @bp.route("/txt", methods=["POST"])
 def txtLogin():
-    # Get domain from form
-    domain = request.form.get("domain").lower().strip().replace("/", "").removesuffix(".")
-    # Get uuid
-    uuid = session["uuid"]
-
-    query = dns.message.make_query(domain, dns.rdatatype.TXT)
-    dns_request = query.to_wire()
-
-    # Send the DNS query over HTTPS
-    response = requests.post('https://hnsdoh.com/dns-query', data=dns_request, headers={'Content-Type': 'application/dns-message'})
-
-    # Parse the DNS response
-    dns_response = dns.message.from_wire(response.content)
-
-    # Loop over TXT records and look for profile avatar
     idns_records = []
-    for record in dns_response.answer:
+    try:
-        if record.rdtype == dns.rdatatype.TXT:
+        # Get domain from form
-            for txt in record:
+        domain = request.form.get("domain").lower().strip().replace("/", "").removesuffix(".")
-                txt_value:str = txt.to_text().strip('"')
+        # Get uuid
-                if txt_value.startswith("IDNS1"):
+        uuid = session["uuid"]
-                    print(txt_value)
+        query = dns.message.make_query(domain, dns.rdatatype.TXT)
-                    idns = txt_value.removeprefix("IDNS1 ")
+        dns_request = query.to_wire()
-                    idns = idns.split(" ")
+        # Send the DNS query over HTTPS
-                    for r in idns:
+        response = requests.post('https://au.hnsdoh.com/dns-query', data=dns_request, headers={'Content-Type': 'application/dns-message'})
-                        idns_records.append(r)
+        # Parse the DNS response
+        dns_response = dns.message.from_wire(response.content)
+    
+        # Loop over TXT records and look for profile
+        for record in dns_response.answer:
+            if record.rdtype == dns.rdatatype.TXT:
+                for txt in record:
+                    txt_value:str = txt.to_text().strip('"')
+                    if txt_value.startswith("IDNS1"):
+                        print(txt_value)
+                        idns = txt_value.removeprefix("IDNS1 ")
+                        idns = idns.split(" ")
+                        for r in idns:
+                            idns_records.append(r)
+    except Exception as e:
+        print(f"Error fetching DNS records: {e}")
+        return render_template("error.html",error="The domain wasn't able to be authenticated.",
+                           message="<br>Double check the TXT record and try again.",
+                           custom="<button onclick='window.location.reload();'>Try again</button>"), 200
+    
+    try:
+        # Get onchain records
+        onchain_response = requests.get(f"https://hsd.hns.au/api/v1/nameresource/{domain}")
+        if onchain_response.status_code == 200:
+            onchain_data = onchain_response.json()
+            if "records" in onchain_data:
+                for record in onchain_data["records"]:
+                    if record["type"] == "TXT":
+                        txt_values = record["txt"]
+                        for txt_value in txt_values:
+                            txt_value = txt_value.strip('"')
+                            if txt_value.startswith("IDNS1"):
+                                print(txt_value)
+                                idns = txt_value.removeprefix("IDNS1 ")
+                                idns = idns.split(" ")
+                                for r in idns:
+                                    idns_records.append(r)
+    except Exception as e:
+        print(f"Error fetching onchain records: {e}")
 
     for record in idns_records:
         print(record)
@@ -623,7 +670,7 @@ def avatar(username):
     dns_request = query.to_wire()
 
     # Send the DNS query over HTTPS
-    response = requests.post('https://hnsdoh.com/dns-query', data=dns_request, headers={'Content-Type': 'application/dns-message'})
+    response = requests.post('https://au.hnsdoh.com/dns-query', data=dns_request, headers={'Content-Type': 'application/dns-message'})
 
     # Parse the DNS response
     dns_response = dns.message.from_wire(response.content)
@@ -640,7 +687,7 @@ def avatar(username):
     
     if avatar_url != "":
         # Download the avatar using DNS-over-HTTPS
-        add_dns_provider("hns", "https://hnsdoh.com/dns-query")
+        add_dns_provider("hns", "https://au.hnsdoh.com/dns-query")
         session = DNSOverHTTPSSession(provider="hns")
         response = session.get(avatar_url)
         with open(f"website/avatars/{username}.png", "wb") as f:

website/varo_auth.py

@@ -12,7 +12,7 @@ def flask_login(request):
 def login(request):
     r = requests.get(f'https://auth.shakestation.io/verify/{request}')
     r = r.json()
-    if r['success'] == False:
+    if not r['success']:
         return False
     
     if 'data' in r: