Creation

.dockerignore

@@ -0,0 +1,4 @@
+.env
+data.db
+__pycache__
+.DS_Store

.env.example

@@ -0,0 +1,4 @@
+BASE_URL="" # Redirect link for when people visit old/dead/non-existant link
+IP_TO_LOCATION="" # "True" or "False" - whether or not you need the IP to location feature,
+                  # If not needed, you can leave the API_KEY blank
+API_KEY="" # API Key for IP2Location.io

.gitignore

@@ -0,0 +1,4 @@
+.env
+data.db
+__pycache__
+.DS_Store

Dockerfile

@@ -0,0 +1,11 @@
+FROM python:3.11-slim
+
+MAINTAINER "parker <mailto:contact@pkrm.dev>"
+
+WORKDIR /
+
+COPY . .
+RUN pip install -r requirements.txt
+
+ENTRYPOINT [ "python" ]
+CMD [ "-u",  "app/linklogger.py" ]

README.md

@@ -0,0 +1,112 @@
+
+# LinkLogger API
+
+A simple API for you to create redirect links on my domain (link.pkrm.dev) and log all IPs that click on the link. Essentially a CLI-only version of Grabify.
+
+Feel free to submit an issue for any problems you experience or if you have an idea for a new feature. If you have a fix for anything, please submit a pull request for review.
+
+## Want to self-host?
+
+#### Bare metal
+Feel free to fork this code and run it yourself, simply install the dependencies, create your `.env` file and run the `linklogger.py` file.
+
+#### Docker
+Use the docker-compose below as an example of running LinkLogger in docker.
+```yaml
+version: '3.3'
+services:
+    linklogger:
+        container_name: linklogger
+        image: packetparker/linklogger
+        ports:
+            - 5252:5252
+        environment:
+            - BASE_URL=https://your.domain
+            - IP_TO_LOCATION=True
+            - API_KEY=Your Key
+        volumes:
+            - /local/file/path:/data
+        restart: unless-stopped
+```
+Variable | Description | Requirement
+---|---|---
+BASE_URL | Redirect link for when people visit old/dead/non-existant link | **Required**
+IP_TO_LOCATION | "True"/"False" Whether or not you want to IP to Location feature (requires IP2Location.io account)  | **Required**
+API_KEY | IP2Location.io API Key | **Required** *unless IP_TO_LOCATION is "False"*
+
+## API Reference
+
+#### Create account/api key
+##### Your account name functions as your API key and will only be provided to you once.
+```http
+GET /signup
+```
+```curl
+curl https://link.pkrm.dev/signup
+```
+
+#### Create new link
+##### Creates a randomized short link that will redirect to the link you provide while logging the IP of the visitor
+```http
+POST /newlink
+```
+```curl
+curl -X POST \
+    -H "Content-type: application/json" \
+    -H "Authorization: Bearer YOUR_ACCOUNT_NAME" \
+    -d '{"redirect_link": "YOUR_LINK_OF_CHOICE"}' \
+    https://link.pkrm.dev/newlink
+```
+
+#### Get all links
+##### Retrieve all of the links and their expiry dates associated with your account
+```curl
+curl -X POST \
+    -H "Authorization: Bearer YOUR_ACCOUNT_NAME" \
+    https://link.pkrm.dev/links
+```
+
+#### Get all logs
+##### Retrieve all IP logs associated with every link on your account
+```http
+POST /records
+```
+```curl
+curl -X POST \
+    -H "Authorization: Bearer YOUR_ACCOUNT_NAME" \
+    https://link.pkrm.dev/records
+```
+
+#### Delete link
+##### Delete the specified link as well as all records associated with it
+```http
+POST /<link>/records
+```
+```curl
+curl -X POST \
+    -H "Authorization: Bearer YOUR_ACCOUNT_NAME" \
+    https://link.pkrm.dev/<link>/delete
+```
+
+#### Renew link
+##### Add 7 more days (from the current date) to the expiry value of the link
+```http
+POST /<link>/Renew
+```
+```curl
+curl -X POST \
+    -H "Authorization: Bearer YOUR_ACCOUNT_NAME" \
+    https://link.pkrm.dev/<link>/renew
+```
+
+#### Link records
+##### Retrieve all IP logs associated with the link
+```http
+POST /<link>/records
+```
+```curl
+curl -X POST \
+    -H "Authorization: Bearer YOUR_ACCOUNT_NAME" \
+    https://link.pkrm.dev/<link>/records
+```
+

app/auth.py

@@ -0,0 +1,16 @@
+from flask_httpauth import HTTPTokenAuth
+import sqlalchemy
+
+from db import engine
+
+
+auth = HTTPTokenAuth(scheme='Bearer')
+
+@auth.verify_token
+def verify_token(token):
+    try:
+        with engine.begin() as conn:
+            token = conn.execute(sqlalchemy.text('SELECT * FROM accounts WHERE account_name = :account_name'), [{'account_name': token}]).fetchone()
+            return token[0]
+    except TypeError:
+        return False

app/db.py

@@ -0,0 +1,31 @@
+import sqlalchemy
+
+engine = sqlalchemy.create_engine('sqlite:///data.db')
+
+def init_db():
+    with engine.begin() as conn:
+        conn.execute(sqlalchemy.text(
+            '''
+            CREATE TABLE IF NOT EXISTS accounts (
+                account_name, PRIMARY KEY (account_name)
+            )
+            '''
+        ))
+        conn.execute(sqlalchemy.text(
+            '''
+            CREATE TABLE IF NOT EXISTS links (
+                owner, link, redirect_link, expire_date,
+                FOREIGN KEY (owner) REFERENCES accounts(account_name), PRIMARY KEY (link)
+            )
+            '''
+        ))
+        conn.execute(sqlalchemy.text(
+            '''
+            CREATE TABLE IF NOT EXISTS records (
+                owner, link, timestamp, ip, location, browser, os, user_agent, isp,
+                FOREIGN KEY (owner) REFERENCES links(owner),
+                FOREIGN KEY (link) REFERENCES links(link))
+            '''
+        ))
+
+        conn.commit()

app/func/delete_link.py

@@ -0,0 +1,20 @@
+import sqlalchemy
+
+from db import engine
+
+"""
+Delete the specified link from the users associated links
+"""
+def delete_link(link, owner):
+    with engine.begin() as conn:
+        try:
+            link_owner = conn.execute(sqlalchemy.text('SELECT owner FROM links WHERE link = :link'), [{'link': link}]).fetchone()[0]
+        except TypeError:
+            return 'Link does not exist', 200
+
+    if owner == link_owner:
+        with engine.begin() as conn:
+            conn.execute(sqlalchemy.text('DELETE FROM links WHERE link = :link'), [{'link': link}])
+            return 'Link has been deleted', 200
+    else:
+        return 'You are not the owner of this link', 401

app/func/link_records.py

@@ -0,0 +1,24 @@
+import sqlalchemy
+import tabulate
+
+from db import engine
+
+"""
+Retrieve all records associated with a specific link
+"""
+def link_records(link, owner):
+    with engine.begin() as conn:
+        try:
+            link_owner = conn.execute(sqlalchemy.text('SELECT owner FROM links WHERE link = :link'), [{'link': link}]).fetchone()[0]
+        except TypeError:
+            return 'Link does not exist', 200
+
+    if owner == link_owner:
+        with engine.begin() as conn:
+            records = conn.execute(sqlalchemy.text('SELECT timestamp, ip, location, browser, os, user_agent, isp FROM records WHERE owner = :owner and link = :link'), [{'owner': owner, 'link': link}]).fetchall()
+        if not records:
+            return 'No records are associated with this link', 200
+    else:
+        return 'You are not the owner of this link', 401
+
+    return tabulate.tabulate(records, headers=['Timestamp', 'IP', 'Location', 'Browser', 'OS', 'User Agent', 'ISP']), 200

app/func/log.py

@@ -0,0 +1,74 @@
+import ip2locationio
+import sqlalchemy
+import datetime
+import validators
+from ua_parser import user_agent_parser
+from dotenv import load_dotenv
+import os
+from ip2locationio.ipgeolocation import IP2LocationIOAPIError
+
+from db import engine
+
+load_dotenv()
+try:
+    ip_to_location = os.getenv('IP_TO_LOCATION').upper().replace('"', '')
+    if ip_to_location == 'TRUE':
+        api_key = os.getenv('API_KEY').replace('"', '')
+    else:
+        api_key = "NO_API_KEY"
+
+    base_url = os.getenv('BASE_URL').replace('"', '')
+# .env File does not exist - likely a docker run
+except AttributeError:
+    ip_to_location = str(os.environ['IP_TO_LOCATION']).upper().replace('"', '')
+    if ip_to_location == 'TRUE':
+        api_key = str(os.environ('API_KEY')).replace('"', '')
+    else:
+        api_key = "NO_API_KEY"
+
+    base_url = str(os.environ('BASE_URL')).replace('"', '')
+
+if not validators.url(base_url):
+    print(base_url)
+    print('BASE_URL varaible is malformed.')
+    exit()
+
+configuration = ip2locationio.Configuration(api_key)
+ipgeolocation = ip2locationio.IPGeolocation(configuration)
+
+"""
+Create a new log record whenever a link is visited
+"""
+def log(link, request):
+    with engine.begin() as conn:
+        try:
+            redirect_link, owner = conn.execute(sqlalchemy.text('SELECT redirect_link, owner FROM links WHERE link = :link'), [{'link': link}]).fetchone()
+        except TypeError:
+            return base_url
+
+    with engine.begin() as conn:
+        if ip_to_location == 'TRUE':
+            # Get IP to GEO via IP2Location.io
+            try:
+                data = ipgeolocation.lookup(request.remote_addr)
+                location = f'{data["country_name"]}, {data["city_name"]}'
+                isp = data['as']
+            # Fatal error, API key is invalid or out of requests, quit
+            except IP2LocationIOAPIError:
+                print('Invalid API key or insifficient credit. Change .env file if you do not need IP to location feature.')
+                location = '-, -'
+                isp = '-'
+        else:
+            location = '-, -'
+            isp = '-'
+
+        timestamp = datetime.datetime.now()
+        ip = request.remote_addr
+        user_agent = request.user_agent.string
+        ua_string = user_agent_parser.Parse(user_agent)
+        browser = ua_string['user_agent']['family']
+        os = f'{ua_string["os"]["family"]} {ua_string["os"]["major"]}'
+
+        conn.execute(sqlalchemy.text('INSERT INTO records (owner, link, timestamp, ip, location, browser, os, user_agent, isp) VALUES (:owner, :link, :timestamp, :ip, :location, :browser, :os, :user_agent, :isp)'), [{'owner': owner, 'link': link, 'timestamp': timestamp, 'ip': ip, 'location': location, 'browser': browser, 'os': os, 'user_agent': user_agent, 'isp': isp}])
+
+    return redirect_link

app/func/newlink.py

@@ -0,0 +1,38 @@
+import validators
+import random
+import string
+import datetime
+import sqlalchemy
+from sqlalchemy import exc
+
+from db import engine
+
+"""
+Generate and return a new randomized link that is connected to the user
+Links are composed of 5 uppercase ASCII characters + numbers
+"""
+def generate_link(request, owner):
+    content_type = request.headers.get('Content-Type')
+    if content_type == 'application/json':
+        try:
+            redirect_link = request.json['redirect_link']
+        except KeyError:
+            return 'Redirect link not provided', 400
+
+        if not validators.url(redirect_link):
+            return 'Redirect link is malformed. Please try again', 400
+    else:
+        return 'Content-Type not supported', 400
+
+    with engine.begin() as conn:
+        choices = string.ascii_uppercase + '1234567890'
+        while True:
+            try:
+                link = ''.join(random.choices(choices, k=5))
+                conn.execute(sqlalchemy.text('INSERT INTO links(owner, link, redirect_link, expire_date) VALUES (:owner, :link, :redirect_link, :expire_date)'), [{'owner': owner, 'link': link, 'redirect_link': redirect_link, 'expire_date': (datetime.datetime.now() + datetime.timedelta(days=7)).strftime('%d/%m/%Y')}])
+                conn.commit()
+                break
+            except exc.IntegrityError:
+                continue
+
+    return link, 200

app/func/renew_link.py

@@ -0,0 +1,22 @@
+import sqlalchemy
+import datetime
+
+from db import engine
+
+"""
+Renew a specified link so that the user can continue logging through that URL
+Adds 7 days from the current date
+"""
+def renew_link(link, owner):
+    with engine.begin() as conn:
+        try:
+            link_owner = conn.execute(sqlalchemy.text('SELECT owner FROM links WHERE link = :link'), [{'link': link}]).fetchone()[0]
+        except TypeError:
+            return 'Link does not exist', 200
+
+    if owner == link_owner:
+        with engine.begin() as conn:
+            conn.execute(sqlalchemy.text('UPDATE links SET expire_date = :expire_date WHERE link = :link'), [{'expire_date': (datetime.datetime.now() + datetime.timedelta(days=7)).strftime('%d/%m/%Y'), 'link': link}])
+            return f'Link renewed, now expires on {(datetime.datetime.now() + datetime.timedelta(days=7)).strftime("%d/%m/%Y")}', 200
+    else:
+        return 'You are not the owner of this link', 401

app/func/signup.py

@@ -0,0 +1,24 @@
+import sqlalchemy
+from sqlalchemy import exc
+import random
+import string
+
+from db import engine
+
+"""
+Generate and return a randomized account string for the user
+Account strings function as API authenticaton keys and are composed
+of 20 uppercase ASCII characters
+"""
+def generate_account():
+    with engine.begin() as conn:
+        while True:
+            try:
+                account_string = ''.join(random.choices(string.ascii_uppercase, k=20))
+                conn.execute(sqlalchemy.text('INSERT INTO accounts(account_name) VALUES(:account_name)'), [{'account_name': account_string}])
+                conn.commit()
+                break
+            except exc.IntegrityError:
+                continue
+
+    return account_string

app/linklogger.py

@@ -0,0 +1,11 @@
+from routes import app
+from db import init_db
+from hypercorn.config import Config
+from hypercorn.asyncio import serve
+import asyncio
+
+if __name__ == '__main__':
+    init_db()
+    config = Config()
+    config.bind =["0.0.0.0:5252"]
+    asyncio.run(serve(app, config))

app/routes.py

@@ -0,0 +1,92 @@
+import flask
+import tabulate
+import sqlalchemy
+import sys
+
+from db import engine
+from auth import auth
+from func.signup import generate_account
+from func.newlink import generate_link
+from func.log import log
+from func.delete_link import delete_link
+from func.renew_link import renew_link
+from func.link_records import link_records
+
+app = flask.Flask(__name__)
+
+
+@app.route('/signup', methods=['GET'])
+def signup():
+    account_name = generate_account()
+    return flask.jsonify({'account_name': account_name})
+
+
+@app.route('/newlink', methods=['POST'])
+@auth.login_required
+def newlink():
+    response = generate_link(flask.request, auth.current_user())
+    return flask.jsonify(msg=response[0]), response[1]
+
+
+"""
+Return all links associated with an account
+"""
+@app.route('/links', methods=['POST'])
+@auth.login_required
+def links():
+    with engine.begin() as conn:
+        links = conn.execute(sqlalchemy.text('SELECT link, expire_date FROM links WHERE owner = :owner'), [{'owner': auth.current_user()}]).fetchall()
+
+    string = ""
+    i = 1
+    for link, expire_date in links:
+        string += f"{i}. {link} - Expires on {expire_date}\n"
+        i += 1
+    return string
+
+
+"""
+Return all records associated with an account, no matter the link
+"""
+@app.route('/records', methods=['POST'])
+@auth.login_required
+def records():
+    with engine.begin() as conn:
+        records = conn.execute(sqlalchemy.text('SELECT timestamp, ip, location, browser, os, user_agent, isp FROM records WHERE owner = :owner'), [{'owner': auth.current_user()}]).fetchall()
+
+    if not records:
+        return flask.jsonify('No records found'), 200
+
+    return tabulate.tabulate(records, headers=['Timestamp', 'IP', 'Location', 'Browser', 'OS', 'User Agent', 'ISP']), 200
+
+
+@app.route('/<link>', methods=['GET'])
+def link(link):
+    redirect_link = log(link, flask.request)
+    return flask.redirect(redirect_link)
+
+
+@app.route('/<link>/delete', methods=['POST'])
+@auth.login_required
+def link_delete(link):
+    response = delete_link(link, auth.current_user())
+    return flask.jsonify(msg=response[0]), response[1]
+
+
+@app.route('/<link>/renew', methods=['POST'])
+@auth.login_required
+def renew_link(link):
+    response = renew_link(link, auth.current_user())
+    return flask.jsonify(msg=response[0]), response[1]
+
+
+@app.route('/<link>/records', methods=['POST'])
+@auth.login_required
+def records_link(link):
+    response = link_records(link, auth.current_user())
+    # If we jsonify the tabulate string it fucks it up, so we have to return
+    # it normally, this check does that
+    if response[0].startswith('Timestamp'):
+        return response[0], response[1]
+    else:
+        return flask.jsonify(msg=response[0]), response[1]

requirements.txt

@@ -0,0 +1,9 @@
+Flask==3.0.0
+Flask-HTTPAuth==4.8.0
+Hypercorn==0.15.0
+ip2location-io==1.0.0
+python-dotenv==1.0.0
+SQLAlchemy==2.0.27
+tabulate==0.9.0
+ua-parser==0.18.0
+validators==0.22.0